346 lines
7.9 KiB
Python
346 lines
7.9 KiB
Python
"""
|
|
Protocol objects to represent objects exposed by different pq implementations.
|
|
"""
|
|
|
|
# Copyright (C) 2020 The Psycopg Team
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any, Protocol, TypeAlias, Union
|
|
from collections.abc import Callable, Sequence
|
|
|
|
from ._enums import Format, Trace
|
|
from .._compat import Self
|
|
|
|
if TYPE_CHECKING:
|
|
from .misc import ConninfoOption, PGnotify, 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 protocol_version(self) -> int: ...
|
|
|
|
@property
|
|
def full_protocol_version(self) -> int: ...
|
|
|
|
@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 used_gssapi(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: ...
|