codekingpro/portable-devtools

1"""2Protocol objects representing different implementations of the same classes.3"""4 5# Copyright (C) 2020 The Psycopg Team6 7from typing import Any, Callable, Generator, Mapping8from typing import List, Optional, Sequence, Tuple, TypeVar, Union9from typing import TYPE_CHECKING10from typing_extensions import TypeAlias11 12from . import pq13from ._enums import PyFormat as PyFormat14from ._compat import Protocol, LiteralString15 16if TYPE_CHECKING:17    from . import sql18    from .rows import Row, RowMaker19    from .pq.abc import PGresult20    from .waiting import Wait, Ready21    from .connection import BaseConnection22    from ._adapters_map import AdaptersMap23 24NoneType: type = type(None)25 26# An object implementing the buffer protocol27Buffer: TypeAlias = Union[bytes, bytearray, memoryview]28 29Query: TypeAlias = Union[LiteralString, bytes, "sql.SQL", "sql.Composed"]30Params: TypeAlias = Union[Sequence[Any], Mapping[str, Any]]31ConnectionType = TypeVar("ConnectionType", bound="BaseConnection[Any]")32PipelineCommand: TypeAlias = Callable[[], None]33DumperKey: TypeAlias = Union[type, Tuple["DumperKey", ...]]34 35# Waiting protocol types36 37RV = TypeVar("RV")38 39PQGenConn: TypeAlias = Generator[Tuple[int, "Wait"], "Ready", RV]40"""Generator for processes where the connection file number can change.41 42This can happen in connection and reset, but not in normal querying.43"""44 45PQGen: TypeAlias = Generator["Wait", "Ready", RV]46"""Generator for processes where the connection file number won't change.47"""48 49 50class WaitFunc(Protocol):51    """52    Wait on the connection which generated `PQgen` and return its final result.53    """54 55    def __call__(56        self, gen: PQGen[RV], fileno: int, timeout: Optional[float] = None57    ) -> RV:58        ...59 60 61# Adaptation types62 63DumpFunc: TypeAlias = Callable[[Any], Buffer]64LoadFunc: TypeAlias = Callable[[Buffer], Any]65 66 67class AdaptContext(Protocol):68    """69    A context describing how types are adapted.70 71    Example of `~AdaptContext` are `~psycopg.Connection`, `~psycopg.Cursor`,72    `~psycopg.adapt.Transformer`, `~psycopg.adapt.AdaptersMap`.73 74    Note that this is a `~typing.Protocol`, so objects implementing75    `!AdaptContext` don't need to explicitly inherit from this class.76 77    """78 79    @property80    def adapters(self) -> "AdaptersMap":81        """The adapters configuration that this object uses."""82        ...83 84    @property85    def connection(self) -> Optional["BaseConnection[Any]"]:86        """The connection used by this object, if available.87 88        :rtype: `~psycopg.Connection` or `~psycopg.AsyncConnection` or `!None`89        """90        ...91 92 93class Dumper(Protocol):94    """95    Convert Python objects of type `!cls` to PostgreSQL representation.96    """97 98    format: pq.Format99    """100    The format that this class `dump()` method produces,101    `~psycopg.pq.Format.TEXT` or `~psycopg.pq.Format.BINARY`.102 103    This is a class attribute.104    """105 106    oid: int107    """The oid to pass to the server, if known; 0 otherwise (class attribute)."""108 109    def __init__(self, cls: type, context: Optional[AdaptContext] = None):110        ...111 112    def dump(self, obj: Any) -> Buffer:113        """Convert the object `!obj` to PostgreSQL representation.114 115        :param obj: the object to convert.116        """117        ...118 119    def quote(self, obj: Any) -> Buffer:120        """Convert the object `!obj` to escaped representation.121 122        :param obj: the object to convert.123        """124        ...125 126    def get_key(self, obj: Any, format: PyFormat) -> DumperKey:127        """Return an alternative key to upgrade the dumper to represent `!obj`.128 129        :param obj: The object to convert130        :param format: The format to convert to131 132        Normally the type of the object is all it takes to define how to dump133        the object to the database. For instance, a Python `~datetime.date` can134        be simply converted into a PostgreSQL :sql:`date`.135 136        In a few cases, just the type is not enough. For example:137 138        - A Python `~datetime.datetime` could be represented as a139          :sql:`timestamptz` or a :sql:`timestamp`, according to whether it140          specifies a `!tzinfo` or not.141 142        - A Python int could be stored as several Postgres types: int2, int4,143          int8, numeric. If a type too small is used, it may result in an144          overflow. If a type too large is used, PostgreSQL may not want to145          cast it to a smaller type.146 147        - Python lists should be dumped according to the type they contain to148          convert them to e.g. array of strings, array of ints (and which149          size of int?...)150 151        In these cases, a dumper can implement `!get_key()` and return a new152        class, or sequence of classes, that can be used to identify the same153        dumper again. If the mechanism is not needed, the method should return154        the same `!cls` object passed in the constructor.155 156        If a dumper implements `get_key()` it should also implement157        `upgrade()`.158 159        """160        ...161 162    def upgrade(self, obj: Any, format: PyFormat) -> "Dumper":163        """Return a new dumper to manage `!obj`.164 165        :param obj: The object to convert166        :param format: The format to convert to167 168        Once `Transformer.get_dumper()` has been notified by `get_key()` that169        this Dumper class cannot handle `!obj` itself, it will invoke170        `!upgrade()`, which should return a new `Dumper` instance, which will171        be reused for every objects for which `!get_key()` returns the same172        result.173        """174        ...175 176 177class Loader(Protocol):178    """179    Convert PostgreSQL values with type OID `!oid` to Python objects.180    """181 182    format: pq.Format183    """184    The format that this class `load()` method can convert,185    `~psycopg.pq.Format.TEXT` or `~psycopg.pq.Format.BINARY`.186 187    This is a class attribute.188    """189 190    def __init__(self, oid: int, context: Optional[AdaptContext] = None):191        ...192 193    def load(self, data: Buffer) -> Any:194        """195        Convert the data returned by the database into a Python object.196 197        :param data: the data to convert.198        """199        ...200 201 202class Transformer(Protocol):203    types: Optional[Tuple[int, ...]]204    formats: Optional[List[pq.Format]]205 206    def __init__(self, context: Optional[AdaptContext] = None):207        ...208 209    @classmethod210    def from_context(cls, context: Optional[AdaptContext]) -> "Transformer":211        ...212 213    @property214    def connection(self) -> Optional["BaseConnection[Any]"]:215        ...216 217    @property218    def encoding(self) -> str:219        ...220 221    @property222    def adapters(self) -> "AdaptersMap":223        ...224 225    @property226    def pgresult(self) -> Optional["PGresult"]:227        ...228 229    def set_pgresult(230        self,231        result: Optional["PGresult"],232        *,233        set_loaders: bool = True,234        format: Optional[pq.Format] = None235    ) -> None:236        ...237 238    def set_dumper_types(self, types: Sequence[int], format: pq.Format) -> None:239        ...240 241    def set_loader_types(self, types: Sequence[int], format: pq.Format) -> None:242        ...243 244    def dump_sequence(245        self, params: Sequence[Any], formats: Sequence[PyFormat]246    ) -> Sequence[Optional[Buffer]]:247        ...248 249    def as_literal(self, obj: Any) -> bytes:250        ...251 252    def get_dumper(self, obj: Any, format: PyFormat) -> Dumper:253        ...254 255    def load_rows(self, row0: int, row1: int, make_row: "RowMaker[Row]") -> List["Row"]:256        ...257 258    def load_row(self, row: int, make_row: "RowMaker[Row]") -> Optional["Row"]:259        ...260 261    def load_sequence(self, record: Sequence[Optional[Buffer]]) -> Tuple[Any, ...]:262        ...263 264    def get_loader(self, oid: int, format: pq.Format) -> Loader:265        ...266