codekingpro/portable-devtools

1"""2Stub representaton of the public objects exposed by the _psycopg module.3 4TODO: this should be generated by mypy's stubgen but it crashes with no5information. Will submit a bug.6"""7 8# Copyright (C) 2020 The Psycopg Team9 10from typing import Any, Iterable, List, Optional, Sequence, Tuple11 12from psycopg import pq13from psycopg import abc14from psycopg.rows import Row, RowMaker15from psycopg.adapt import AdaptersMap, PyFormat16from psycopg.pq.abc import PGconn, PGresult17from psycopg.connection import BaseConnection18from psycopg._compat import Deque19 20class Transformer(abc.AdaptContext):21    types: Optional[Tuple[int, ...]]22    formats: Optional[List[pq.Format]]23    def __init__(self, context: Optional[abc.AdaptContext] = None): ...24    @classmethod25    def from_context(cls, context: Optional[abc.AdaptContext]) -> "Transformer": ...26    @property27    def connection(self) -> Optional[BaseConnection[Any]]: ...28    @property29    def encoding(self) -> str: ...30    @property31    def adapters(self) -> AdaptersMap: ...32    @property33    def pgresult(self) -> Optional[PGresult]: ...34    def set_pgresult(35        self,36        result: Optional["PGresult"],37        *,38        set_loaders: bool = True,39        format: Optional[pq.Format] = None,40    ) -> None: ...41    def set_dumper_types(self, types: Sequence[int], format: pq.Format) -> None: ...42    def set_loader_types(self, types: Sequence[int], format: pq.Format) -> None: ...43    def dump_sequence(44        self, params: Sequence[Any], formats: Sequence[PyFormat]45    ) -> Sequence[Optional[abc.Buffer]]: ...46    def as_literal(self, obj: Any) -> bytes: ...47    def get_dumper(self, obj: Any, format: PyFormat) -> abc.Dumper: ...48    def load_rows(self, row0: int, row1: int, make_row: RowMaker[Row]) -> List[Row]: ...49    def load_row(self, row: int, make_row: RowMaker[Row]) -> Optional[Row]: ...50    def load_sequence(51        self, record: Sequence[Optional[abc.Buffer]]52    ) -> Tuple[Any, ...]: ...53    def get_loader(self, oid: int, format: pq.Format) -> abc.Loader: ...54 55# Generators56def connect(conninfo: str) -> abc.PQGenConn[PGconn]: ...57def execute(pgconn: PGconn) -> abc.PQGen[List[PGresult]]: ...58def send(pgconn: PGconn) -> abc.PQGen[None]: ...59def fetch_many(pgconn: PGconn) -> abc.PQGen[List[PGresult]]: ...60def fetch(pgconn: PGconn) -> abc.PQGen[Optional[PGresult]]: ...61def pipeline_communicate(62    pgconn: PGconn, commands: Deque[abc.PipelineCommand]63) -> abc.PQGen[List[List[PGresult]]]: ...64def wait_c(65    gen: abc.PQGen[abc.RV], fileno: int, timeout: Optional[float] = None66) -> abc.RV: ...67 68# Copy support69def format_row_text(70    row: Sequence[Any], tx: abc.Transformer, out: Optional[bytearray] = None71) -> bytearray: ...72def format_row_binary(73    row: Sequence[Any], tx: abc.Transformer, out: Optional[bytearray] = None74) -> bytearray: ...75def parse_row_text(data: abc.Buffer, tx: abc.Transformer) -> Tuple[Any, ...]: ...76def parse_row_binary(data: abc.Buffer, tx: abc.Transformer) -> Tuple[Any, ...]: ...77 78# Arrays optimization79def array_load_text(80    data: abc.Buffer, loader: abc.Loader, delimiter: bytes = b","81) -> List[Any]: ...82def array_load_binary(data: abc.Buffer, tx: abc.Transformer) -> List[Any]: ...83 84# vim: set syntax=python:85