Method _exec_command

psycopg/psycopg/_connection_base.py:445–489

Generator to send a command and receive the result to the backend. Only used to implement internal commands such as "commit", with eventual arguments bound client-side. The cursor can do more complex stuff.

(
        self, command: QueryNoTemplate, result_format: pq.Format = TEXT
    )

        return conn

    def _exec_command(
        self, command: QueryNoTemplate, result_format: pq.Format = TEXT
    ) -> PQGen[PGresult | None]:
        """
        Generator to send a command and receive the result to the backend.

        Only used to implement internal commands such as "commit", with eventual
        arguments bound client-side. The cursor can do more complex stuff.
        """
        self._check_connection_ok()

        if isinstance(command, str):
            command = command.encode(self.pgconn._encoding)
        elif isinstance(command, Composable):
            command = command.as_bytes(self)

        if self._pipeline:
            cmd = partial(
                self.pgconn.send_query_params,
                command,
                None,
                result_format=result_format,
            )
            self._pipeline.command_queue.append(cmd)
            self._pipeline.result_queue.append(None)
            return None

        # Unless needed, use the simple query protocol, e.g. to interact with
        # pgbouncer. In pipeline mode we always use the advanced query protocol
        # instead, see #350
        if result_format == TEXT:
            self.pgconn.send_query(command)
        else:
            self.pgconn.send_query_params(command, None, result_format=result_format)

        result: PGresult = (yield from generators.execute(self.pgconn))[-1]
        if result.status != COMMAND_OK and result.status != TUPLES_OK:
            if result.status == FATAL_ERROR:
                raise e.error_from_result(result, encoding=self.pgconn._encoding)
            else:
                raise e.InterfaceError(
                    f"unexpected result {pq.ExecStatus(result.status).name}"
                    f" from command {command.decode()!r}"
                )
        return result

    def _deallocate(self, name: bytes | None) -> PQGen[None]:
        """