codekingpro/portable-devtools

1"""2psycopg client-side binding cursors3"""4 5# Copyright (C) 2022 The Psycopg Team6 7from typing import Optional, Tuple, TYPE_CHECKING8from functools import partial9 10from ._queries import PostgresQuery, PostgresClientQuery11 12from . import pq13from . import adapt14from . import errors as e15from .abc import ConnectionType, Query, Params16from .rows import Row17from .cursor import BaseCursor, Cursor18from ._preparing import Prepare19from .cursor_async import AsyncCursor20 21if TYPE_CHECKING:22    from typing import Any  # noqa: F40123    from .connection import Connection  # noqa: F40124    from .connection_async import AsyncConnection  # noqa: F40125 26TEXT = pq.Format.TEXT27BINARY = pq.Format.BINARY28 29 30class ClientCursorMixin(BaseCursor[ConnectionType, Row]):31    def mogrify(self, query: Query, params: Optional[Params] = None) -> str:32        """33        Return the query and parameters merged.34 35        Parameters are adapted and merged to the query the same way that36        `!execute()` would do.37 38        """39        self._tx = adapt.Transformer(self)40        pgq = self._convert_query(query, params)41        return pgq.query.decode(self._tx.encoding)42 43    def _execute_send(44        self,45        query: PostgresQuery,46        *,47        force_extended: bool = False,48        binary: Optional[bool] = None,49    ) -> None:50        if binary is None:51            fmt = self.format52        else:53            fmt = BINARY if binary else TEXT54 55        if fmt == BINARY:56            raise e.NotSupportedError(57                "client-side cursors don't support binary results"58            )59 60        self._query = query61 62        if self._conn._pipeline:63            # In pipeline mode always use PQsendQueryParams - see #31464            # Multiple statements in the same query are not allowed anyway.65            self._conn._pipeline.command_queue.append(66                partial(self._pgconn.send_query_params, query.query, None)67            )68        elif force_extended:69            self._pgconn.send_query_params(query.query, None)70        else:71            # If we can, let's use simple query protocol,72            # as it can execute more than one statement in a single query.73            self._pgconn.send_query(query.query)74 75    def _convert_query(76        self, query: Query, params: Optional[Params] = None77    ) -> PostgresQuery:78        pgq = PostgresClientQuery(self._tx)79        pgq.convert(query, params)80        return pgq81 82    def _get_prepared(83        self, pgq: PostgresQuery, prepare: Optional[bool] = None84    ) -> Tuple[Prepare, bytes]:85        return (Prepare.NO, b"")86 87 88class ClientCursor(ClientCursorMixin["Connection[Any]", Row], Cursor[Row]):89    __module__ = "psycopg"90 91 92class AsyncClientCursor(93    ClientCursorMixin["AsyncConnection[Any]", Row], AsyncCursor[Row]94):95    __module__ = "psycopg"96