""" Protocol objects to represent objects exposed by different pq implementations. """ # Copyright (C) 2020 The Psycopg Team from __future__ import annotations from typing import Any, Callable, Protocol, Sequence, TYPE_CHECKING from typing import Union # drop with Python 3.8 from ._enums import Format, Trace from .._compat import Self, TypeAlias if TYPE_CHECKING: from .misc import PGnotify, ConninfoOption, PGresAttDesc # An object implementing the buffer protocol (ish) Buffer: TypeAlias = Union[bytes, bytearray, memoryview] class PGconn(Protocol): notice_handler: Callable[[PGresult], None] | None notify_handler: Callable[[PGnotify], None] | None @classmethod def connect(cls, conninfo: bytes) -> Self: ... @classmethod def connect_start(cls, conninfo: bytes) -> Self: ... def connect_poll(self) -> int: ... def finish(self) -> None: ... @property def info(self) -> list[ConninfoOption]: ... def reset(self) -> None: ... def reset_start(self) -> None: ... def reset_poll(self) -> int: ... @classmethod def ping(self, conninfo: bytes) -> int: ... @property def db(self) -> bytes: ... @property def user(self) -> bytes: ... @property def password(self) -> bytes: ... @property def host(self) -> bytes: ... @property def hostaddr(self) -> bytes: ... @property def port(self) -> bytes: ... @property def tty(self) -> bytes: ... @property def options(self) -> bytes: ... @property def status(self) -> int: ... @property def transaction_status(self) -> int: ... def parameter_status(self, name: bytes) -> bytes | None: ... @property def error_message(self) -> bytes: ... def get_error_message(self, encoding: str = ...) -> str: ... @property def _encoding(self) -> str: ... @property def server_version(self) -> int: ... @property def socket(self) -> int: ... @property def backend_pid(self) -> int: ... @property def needs_password(self) -> bool: ... @property def used_password(self) -> bool: ... @property def ssl_in_use(self) -> bool: ... def exec_(self, command: bytes) -> PGresult: ... def send_query(self, command: bytes) -> None: ... def exec_params( self, command: bytes, param_values: Sequence[Buffer | None] | None, param_types: Sequence[int] | None = None, param_formats: Sequence[int] | None = None, result_format: int = Format.TEXT, ) -> PGresult: ... def send_query_params( self, command: bytes, param_values: Sequence[Buffer | None] | None, param_types: Sequence[int] | None = None, param_formats: Sequence[int] | None = None, result_format: int = Format.TEXT, ) -> None: ... def send_prepare( self, name: bytes, command: bytes, param_types: Sequence[int] | None = None, ) -> None: ... def send_query_prepared( self, name: bytes, param_values: Sequence[Buffer | None] | None, param_formats: Sequence[int] | None = None, result_format: int = Format.TEXT, ) -> None: ... def prepare( self, name: bytes, command: bytes, param_types: Sequence[int] | None = None, ) -> PGresult: ... def exec_prepared( self, name: bytes, param_values: Sequence[Buffer] | None, param_formats: Sequence[int] | None = None, result_format: int = 0, ) -> PGresult: ... def describe_prepared(self, name: bytes) -> PGresult: ... def send_describe_prepared(self, name: bytes) -> None: ... def describe_portal(self, name: bytes) -> PGresult: ... def send_describe_portal(self, name: bytes) -> None: ... def close_prepared(self, name: bytes) -> PGresult: ... def send_close_prepared(self, name: bytes) -> None: ... def close_portal(self, name: bytes) -> PGresult: ... def send_close_portal(self, name: bytes) -> None: ... def get_result(self) -> PGresult | None: ... def consume_input(self) -> None: ... def is_busy(self) -> int: ... @property def nonblocking(self) -> int: ... @nonblocking.setter def nonblocking(self, arg: int) -> None: ... def flush(self) -> int: ... def set_single_row_mode(self) -> None: ... def set_chunked_rows_mode(self, size: int) -> None: ... def cancel_conn(self) -> PGcancelConn: ... def get_cancel(self) -> PGcancel: ... def notifies(self) -> PGnotify | None: ... def put_copy_data(self, buffer: Buffer) -> int: ... def put_copy_end(self, error: bytes | None = None) -> int: ... def get_copy_data(self, async_: int) -> tuple[int, memoryview]: ... def trace(self, fileno: int) -> None: ... def set_trace_flags(self, flags: Trace) -> None: ... def untrace(self) -> None: ... def encrypt_password( self, passwd: bytes, user: bytes, algorithm: bytes | None = None ) -> bytes: ... def change_password(self, user: bytes, passwd: bytes) -> None: ... def make_empty_result(self, exec_status: int) -> PGresult: ... @property def pipeline_status(self) -> int: ... def enter_pipeline_mode(self) -> None: ... def exit_pipeline_mode(self) -> None: ... def pipeline_sync(self) -> None: ... def send_flush_request(self) -> None: ... class PGresult(Protocol): def clear(self) -> None: ... @property def status(self) -> int: ... @property def error_message(self) -> bytes: ... def get_error_message(self, encoding: str = ...) -> str: ... def error_field(self, fieldcode: int) -> bytes | None: ... @property def ntuples(self) -> int: ... @property def nfields(self) -> int: ... def fname(self, column_number: int) -> bytes | None: ... def ftable(self, column_number: int) -> int: ... def ftablecol(self, column_number: int) -> int: ... def fformat(self, column_number: int) -> int: ... def ftype(self, column_number: int) -> int: ... def fmod(self, column_number: int) -> int: ... def fsize(self, column_number: int) -> int: ... @property def binary_tuples(self) -> int: ... def get_value(self, row_number: int, column_number: int) -> bytes | None: ... @property def nparams(self) -> int: ... def param_type(self, param_number: int) -> int: ... @property def command_status(self) -> bytes | None: ... @property def command_tuples(self) -> int | None: ... @property def oid_value(self) -> int: ... def set_attributes(self, descriptions: list[PGresAttDesc]) -> None: ... class PGcancelConn(Protocol): def start(self) -> None: ... def blocking(self) -> None: ... def poll(self) -> int: ... @property def status(self) -> int: ... @property def socket(self) -> int: ... @property def error_message(self) -> bytes: ... def get_error_message(self, encoding: str = ...) -> str: ... def reset(self) -> None: ... def finish(self) -> None: ... class PGcancel(Protocol): def free(self) -> None: ... def cancel(self) -> None: ... class Conninfo(Protocol): @classmethod def get_defaults(cls) -> list[ConninfoOption]: ... @classmethod def parse(cls, conninfo: bytes) -> list[ConninfoOption]: ... @classmethod def _options_from_array(cls, opts: Sequence[Any]) -> list[ConninfoOption]: ... class Escaping(Protocol): def __init__(self, conn: PGconn | None = None): ... def escape_literal(self, data: Buffer) -> bytes: ... def escape_identifier(self, data: Buffer) -> bytes: ... def escape_string(self, data: Buffer) -> bytes: ... def escape_bytea(self, data: Buffer) -> bytes: ... def unescape_bytea(self, data: Buffer) -> bytes: ...