psycopg2: Add missing modules, add annotations (#10630)

This commit is contained in:
Ali Hamdan
2023-09-13 12:11:54 +02:00
committed by GitHub
parent c9bf034dd4
commit 71a35b4343
16 changed files with 589 additions and 353 deletions

View File

@@ -1,5 +1,3 @@
psycopg2.connection
psycopg2.cursor
psycopg2.pool.AbstractConnectionPool.closeall
psycopg2.pool.AbstractConnectionPool.getconn
psycopg2.pool.AbstractConnectionPool.putconn
psycopg2.pool.AbstractConnectionPool.(closeall|getconn|putconn)
psycopg2.(extras|_range).RangeAdapter.name
psycopg2.(_psycopg|extensions).connection.async # async is a reserved keyword in Python 3.7

View File

@@ -0,0 +1,40 @@
from __future__ import annotations
from typing_extensions import assert_type
import psycopg2
from psycopg2.extensions import connection, cursor
class MyCursor(cursor):
pass
class MyConnection(connection):
pass
def custom_connection(dsn: str) -> MyConnection:
return MyConnection(dsn, async_=0)
# -> psycopg2.extensions.connection
assert_type(psycopg2.connect(), connection)
assert_type(psycopg2.connect("test-conn"), connection)
assert_type(psycopg2.connect(None), connection)
assert_type(psycopg2.connect("test-conn", connection_factory=None), connection)
assert_type(psycopg2.connect(cursor_factory=MyCursor), connection)
assert_type(psycopg2.connect("test-conn", cursor_factory=MyCursor), connection)
assert_type(psycopg2.connect(None, cursor_factory=MyCursor), connection)
assert_type(psycopg2.connect("test-conn", connection_factory=None, cursor_factory=MyCursor), connection)
# -> custom_connection
assert_type(psycopg2.connect(connection_factory=MyConnection), MyConnection)
assert_type(psycopg2.connect("test-conn", connection_factory=MyConnection), MyConnection)
assert_type(psycopg2.connect("test-conn", MyConnection), MyConnection)
assert_type(psycopg2.connect(connection_factory=custom_connection), MyConnection)
assert_type(psycopg2.connect(connection_factory=MyConnection, cursor_factory=MyCursor), MyConnection)
assert_type(psycopg2.connect("test-conn", connection_factory=MyConnection, cursor_factory=MyCursor), MyConnection)
assert_type(psycopg2.connect(connection_factory=custom_connection, cursor_factory=MyCursor), MyConnection)

View File

@@ -0,0 +1,56 @@
from __future__ import annotations
from typing_extensions import assert_type
import psycopg2.extensions
import psycopg2.extras
from psycopg2.extensions import make_dsn
# make_dsn
# --------
# (None) -> str
assert_type(make_dsn(), str)
assert_type(make_dsn(None), str)
assert_type(make_dsn(dsn=None), str)
# (bytes) -> bytes
assert_type(make_dsn(b""), bytes)
assert_type(make_dsn(dsn=b""), bytes)
# (bytes, **Kwargs) -> str
assert_type(make_dsn(b"", database=""), str)
assert_type(make_dsn(dsn=b"", database=""), str)
# (str, **OptionalKwargs) -> str
assert_type(make_dsn(""), str)
assert_type(make_dsn(dsn=""), str)
assert_type(make_dsn("", database=None), str)
assert_type(make_dsn(dsn="", database=None), str)
# connection.cursor
# -----------------
# (name?, None?, ...) -> psycopg2.extensions.cursor
conn = psycopg2.connect("test-conn")
assert_type(conn.cursor(), psycopg2.extensions.cursor)
assert_type(conn.cursor("test-cur"), psycopg2.extensions.cursor)
assert_type(conn.cursor("test-cur", None), psycopg2.extensions.cursor)
assert_type(conn.cursor("test-cur", cursor_factory=None), psycopg2.extensions.cursor)
# (name?, cursor_factory(), ...) -> custom_cursor
class MyCursor(psycopg2.extensions.cursor):
pass
assert_type(conn.cursor("test-cur", cursor_factory=MyCursor), MyCursor)
assert_type(conn.cursor("test-cur", cursor_factory=lambda c, n: MyCursor(c, n)), MyCursor)
dconn = psycopg2.extras.DictConnection("test-dconn")
assert_type(dconn.cursor(), psycopg2.extras.DictCursor)
assert_type(dconn.cursor("test-dcur"), psycopg2.extras.DictCursor)
assert_type(dconn.cursor("test-dcur", None), psycopg2.extras.DictCursor)
assert_type(dconn.cursor("test-dcur", cursor_factory=None), psycopg2.extras.DictCursor)
assert_type(dconn.cursor("test-dcur", cursor_factory=MyCursor), MyCursor)

View File

@@ -3,4 +3,4 @@ upstream_repository = "https://github.com/psycopg/psycopg2"
partial_stub = true
[tool.stubtest]
ignore_missing_stub = true
ignore_missing_stub = false

View File

@@ -27,8 +27,8 @@ from psycopg2._psycopg import (
Warning as Warning,
__libpq_version__ as __libpq_version__,
apilevel as apilevel,
connection as connection,
cursor as cursor,
connection,
cursor,
paramstyle as paramstyle,
threadsafety as threadsafety,
)
@@ -36,10 +36,19 @@ from psycopg2._psycopg import (
_T_conn = TypeVar("_T_conn", bound=connection)
@overload
def connect(dsn: str, connection_factory: Callable[..., _T_conn], cursor_factory: None = None, **kwargs: Any) -> _T_conn: ...
def connect(
dsn: str | None,
connection_factory: Callable[..., _T_conn],
cursor_factory: Callable[..., cursor] | None = None,
**kwargs: Any,
) -> _T_conn: ...
@overload
def connect(
dsn: str | None = None, *, connection_factory: Callable[..., _T_conn], cursor_factory: None = None, **kwargs: Any
dsn: str | None = None,
*,
connection_factory: Callable[..., _T_conn],
cursor_factory: Callable[..., cursor] | None = None,
**kwargs: Any,
) -> _T_conn: ...
@overload
def connect(

View File

@@ -1,9 +1,9 @@
from _typeshed import Incomplete
from typing import Any
import ipaddress as ipaddress
from _typeshed import Unused
ipaddress: Any
from psycopg2._psycopg import QuotedString, connection, cursor
def register_ipaddress(conn_or_curs: Incomplete | None = None) -> None: ...
def cast_interface(s, cur: Incomplete | None = None): ...
def cast_network(s, cur: Incomplete | None = None): ...
def adapt_ipaddress(obj): ...
def register_ipaddress(conn_or_curs: connection | cursor | None = None) -> None: ...
def cast_interface(s: str, cur: Unused = None) -> ipaddress.IPv4Interface | ipaddress.IPv6Interface | None: ...
def cast_network(s: str, cur: Unused = None) -> ipaddress.IPv4Network | ipaddress.IPv6Network | None: ...
def adapt_ipaddress(obj: object) -> QuotedString: ...

View File

@@ -1,5 +1,8 @@
from _typeshed import Incomplete
from collections.abc import Callable
from typing import Any
from typing_extensions import Self
from psycopg2._psycopg import _type, connection, cursor
JSON_OID: int
JSONARRAY_OID: int
@@ -8,19 +11,23 @@ JSONBARRAY_OID: int
class Json:
adapted: Any
def __init__(self, adapted, dumps: Incomplete | None = None) -> None: ...
def __conform__(self, proto): ...
def dumps(self, obj): ...
def prepare(self, conn) -> None: ...
def getquoted(self): ...
def __init__(self, adapted: Any, dumps: Callable[..., str] | None = None) -> None: ...
def __conform__(self, proto) -> Self | None: ...
def dumps(self, obj: Any) -> str: ...
def prepare(self, conn: connection | None) -> None: ...
def getquoted(self) -> bytes: ...
def register_json(
conn_or_curs: Incomplete | None = None,
conn_or_curs: connection | cursor | None = None,
globally: bool = False,
loads: Incomplete | None = None,
oid: Incomplete | None = None,
array_oid: Incomplete | None = None,
loads: Callable[..., Any] | None = None,
oid: int | None = None,
array_oid: int | None = None,
name: str = "json",
): ...
def register_default_json(conn_or_curs: Incomplete | None = None, globally: bool = False, loads: Incomplete | None = None): ...
def register_default_jsonb(conn_or_curs: Incomplete | None = None, globally: bool = False, loads: Incomplete | None = None): ...
) -> tuple[_type, _type | None]: ...
def register_default_json(
conn_or_curs: connection | cursor | None = None, globally: bool = False, loads: Callable[..., Any] | None = None
) -> tuple[_type, _type | None]: ...
def register_default_jsonb(
conn_or_curs: connection | cursor | None = None, globally: bool = False, loads: Callable[..., Any] | None = None
) -> tuple[_type, _type | None]: ...

View File

@@ -1,109 +1,148 @@
import datetime as dt
from _typeshed import Incomplete, ReadableBuffer, SupportsRead, SupportsReadline, SupportsTrunc, SupportsWrite, Unused
from collections.abc import Callable, Iterable, Mapping, Sequence
from types import TracebackType
from typing import Any, TypeVar, overload
from typing_extensions import Literal, Self, TypeAlias
from typing import Any, NoReturn, Protocol, SupportsInt, TypeVar, overload, type_check_only
from typing_extensions import Literal, Self, SupportsIndex, TypeAlias
import psycopg2
import psycopg2.extensions
from psycopg2.sql import Composable
_Vars: TypeAlias = Sequence[Any] | Mapping[str, Any] | None
_AcceptedByInt: TypeAlias = str | ReadableBuffer | SupportsInt | SupportsIndex | SupportsTrunc
@type_check_only
class _type:
# The class doesn't exist at runtime but following attributes have type "psycopg2._psycopg.type"
name: str
values: tuple[int, ...]
def __call__(self, __value: str | bytes | None, __cur: cursor | None) -> Any: ...
BINARY: _type
BINARYARRAY: _type
BOOLEAN: _type
BOOLEANARRAY: _type
BYTES: _type
BYTESARRAY: _type
CIDRARRAY: _type
DATE: _type
DATEARRAY: _type
DATETIME: _type
DATETIMEARRAY: _type
DATETIMETZ: _type
DATETIMETZARRAY: _type
DECIMAL: _type
DECIMALARRAY: _type
FLOAT: _type
FLOATARRAY: _type
INETARRAY: _type
INTEGER: _type
INTEGERARRAY: _type
INTERVAL: _type
INTERVALARRAY: _type
LONGINTEGER: _type
LONGINTEGERARRAY: _type
MACADDRARRAY: _type
NUMBER: _type
PYDATE: _type
PYDATEARRAY: _type
PYDATETIME: _type
PYDATETIMEARRAY: _type
PYDATETIMETZ: _type
PYDATETIMETZARRAY: _type
PYINTERVAL: _type
PYINTERVALARRAY: _type
PYTIME: _type
PYTIMEARRAY: _type
ROWID: _type
ROWIDARRAY: _type
STRING: _type
STRINGARRAY: _type
TIME: _type
TIMEARRAY: _type
UNICODE: _type
UNICODEARRAY: _type
UNKNOWN: _type
BINARY: Any
BINARYARRAY: Any
BOOLEAN: Any
BOOLEANARRAY: Any
BYTES: Any
BYTESARRAY: Any
CIDRARRAY: Any
DATE: Any
DATEARRAY: Any
DATETIME: Any
DATETIMEARRAY: Any
DATETIMETZ: Any
DATETIMETZARRAY: Any
DECIMAL: Any
DECIMALARRAY: Any
FLOAT: Any
FLOATARRAY: Any
INETARRAY: Any
INTEGER: Any
INTEGERARRAY: Any
INTERVAL: Any
INTERVALARRAY: Any
LONGINTEGER: Any
LONGINTEGERARRAY: Any
MACADDRARRAY: Any
NUMBER: Any
PYDATE: Any
PYDATEARRAY: Any
PYDATETIME: Any
PYDATETIMEARRAY: Any
PYDATETIMETZ: Any
PYDATETIMETZARRAY: Any
PYINTERVAL: Any
PYINTERVALARRAY: Any
PYTIME: Any
PYTIMEARRAY: Any
REPLICATION_LOGICAL: int
REPLICATION_PHYSICAL: int
ROWID: Any
ROWIDARRAY: Any
STRING: Any
STRINGARRAY: Any
TIME: Any
TIMEARRAY: Any
UNICODE: Any
UNICODEARRAY: Any
UNKNOWN: Any
adapters: dict[Any, Any]
class _ISQLQuoteProto(Protocol):
# Objects conforming this protocol should implement a getquoted() and optionally a prepare() method.
# The real ISQLQuote class is implemented below with more stuff.
def getquoted(self) -> bytes: ...
# def prepare(self, __conn: connection) -> None: ... # optional
adapters: dict[tuple[type[Any], type[ISQLQuote]], Callable[[Any], _ISQLQuoteProto]]
apilevel: str
binary_types: dict[Any, Any]
encodings: dict[Any, Any]
encodings: dict[str, str]
paramstyle: str
sqlstate_errors: dict[Any, Any]
string_types: dict[Any, Any]
sqlstate_errors: dict[str, type[Error]]
string_types: dict[int, _type]
threadsafety: int
__libpq_version__: int
class _SupportsReadAndReadline(SupportsRead[str], SupportsReadline[str]): ...
class _SupportsReadAndReadlineAndWrite(_SupportsReadAndReadline, SupportsWrite[str]): ...
class cursor:
arraysize: int
binary_types: Any
closed: Any
connection: Any
description: Any
itersize: Any
lastrowid: Any
name: Any
pgresult_ptr: Any
query: Any
row_factory: Any
rowcount: int
rownumber: int
binary_types: Incomplete | None
connection: _Connection
itersize: int
row_factory: Incomplete | None
scrollable: bool | None
statusmessage: Any
string_types: Any
typecaster: Any
tzinfo_factory: Any
string_types: Incomplete | None
tzinfo_factory: Callable[..., dt.tzinfo]
withhold: bool
def __init__(self, conn: connection, name: str | bytes | None = ...) -> None: ...
def callproc(self, procname, parameters=...): ...
def cast(self, oid, s): ...
def close(self): ...
def copy_expert(self, sql: str | bytes | Composable, file, size=...): ...
def copy_from(self, file, table, sep=..., null=..., size=..., columns=...): ...
def copy_to(self, file, table, sep=..., null=..., columns=...): ...
def execute(self, query: str | bytes | Composable, vars: _Vars = ...) -> None: ...
def __init__(self, conn: _Connection, name: str | bytes | None = None) -> None: ...
@property
def closed(self) -> bool: ...
@property
def lastrowid(self) -> int: ...
@property
def name(self) -> Incomplete | None: ...
@property
def query(self) -> bytes | None: ...
@property
def description(self) -> tuple[Column, ...] | None: ...
@property
def rowcount(self) -> int: ...
@property
def rownumber(self) -> int: ...
@property
def typecaster(self) -> Incomplete | None: ...
@property
def statusmessage(self) -> str | None: ...
@property
def pgresult_ptr(self) -> int | None: ...
def callproc(self, __procname: str | bytes, __parameters: _Vars = None) -> None: ...
def cast(self, __oid: int, __s: str | bytes) -> Any: ...
def close(self) -> None: ...
def copy_expert(self, sql: str | bytes | Composable, file: _SupportsReadAndReadlineAndWrite, size: int = 8192) -> None: ...
def copy_from(
self,
file: _SupportsReadAndReadline,
table: str,
sep: str = "\t",
null: str = "\\N",
size: int = 8192,
columns: Iterable[str] | None = None,
) -> None: ...
def copy_to(
self, file: SupportsWrite[str], table: str, sep: str = "\t", null: str = "\\N", columns: Iterable[str] | None = None
) -> None: ...
def execute(self, query: str | bytes | Composable, vars: _Vars = None) -> None: ...
def executemany(self, query: str | bytes | Composable, vars_list: Iterable[_Vars]) -> None: ...
def fetchall(self) -> list[tuple[Any, ...]]: ...
def fetchmany(self, size: int | None = ...) -> list[tuple[Any, ...]]: ...
def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]: ...
def fetchone(self) -> tuple[Any, ...] | None: ...
def mogrify(self, *args, **kwargs): ...
def nextset(self): ...
def scroll(self, value, mode=...): ...
def setinputsizes(self, sizes): ...
def setoutputsize(self, size, column=...): ...
def mogrify(self, query: str | bytes, vars: _Vars | None = None) -> bytes: ...
def nextset(self) -> NoReturn: ... # not supported
def scroll(self, value: int, mode: Literal["absolute", "relative"] = "relative") -> None: ...
def setinputsizes(self, sizes: Unused) -> None: ...
def setoutputsize(self, __size: int, __column: int = ...) -> None: ...
def __enter__(self) -> Self: ...
def __exit__(
self, type: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None
@@ -114,24 +153,28 @@ class cursor:
_Cursor: TypeAlias = cursor
class AsIs:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
def __init__(self, __obj: object, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
class Binary:
adapted: Any
buffer: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def prepare(self, conn): ...
def __conform__(self, *args, **kwargs): ...
def __init__(self, __str: object, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
@property
def buffer(self) -> Any: ...
def getquoted(self) -> bytes: ...
def prepare(self, __conn: connection) -> None: ...
def __conform__(self, __proto) -> Self | None: ...
class Boolean:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
def __init__(self, __obj: object, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
class Column:
display_size: Any
@@ -214,14 +257,40 @@ class ConnectionInfo:
def parameter_status(self, name: str) -> str | None: ...
def ssl_attribute(self, name: str) -> str | None: ...
class DataError(psycopg2.DatabaseError): ...
class DatabaseError(psycopg2.Error): ...
class Error(Exception):
cursor: _Cursor | None
diag: Diagnostics
pgcode: str | None
pgerror: str | None
def __init__(self, *args, **kwargs) -> None: ...
def __reduce__(self): ...
def __setstate__(self, state): ...
class DatabaseError(Error): ...
class DataError(DatabaseError): ...
class IntegrityError(DatabaseError): ...
class InternalError(DatabaseError): ...
class NotSupportedError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...
class QueryCanceledError(OperationalError): ...
class TransactionRollbackError(OperationalError): ...
class InterfaceError(Error): ...
class Warning(Exception): ...
class ISQLQuote:
_wrapped: Any
def __init__(self, __wrapped: object, **kwargs) -> None: ...
def getbinary(self) -> Incomplete: ...
def getbuffer(self) -> Incomplete: ...
def getquoted(self) -> bytes: ...
class Decimal:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
def __init__(self, __value: object, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
class Diagnostics:
column_name: str | None
@@ -244,46 +313,27 @@ class Diagnostics:
table_name: str | None
def __init__(self, __err: Error) -> None: ...
class Error(Exception):
cursor: _Cursor | None
diag: Diagnostics
pgcode: str | None
pgerror: str | None
def __init__(self, *args, **kwargs) -> None: ...
def __reduce__(self): ...
def __setstate__(self, state): ...
class Float:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
class ISQLQuote:
_wrapped: Any
def __init__(self, *args, **kwargs) -> None: ...
def getbinary(self, *args, **kwargs): ...
def getbuffer(self, *args, **kwargs): ...
def getquoted(self, *args, **kwargs): ...
def __init__(self, __value: float, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> float: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
class Int:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
class IntegrityError(psycopg2.DatabaseError): ...
class InterfaceError(psycopg2.Error): ...
class InternalError(psycopg2.DatabaseError): ...
def __init__(self, __value: _AcceptedByInt, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
class List:
adapted: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def prepare(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
class NotSupportedError(psycopg2.DatabaseError): ...
def __init__(self, __objs: list[object], **kwargs: Unused) -> None: ...
@property
def adapted(self) -> list[Any]: ...
def getquoted(self) -> bytes: ...
def prepare(self, __conn: connection) -> None: ...
def __conform__(self, __proto) -> Self | None: ...
class Notify:
channel: Any
@@ -300,27 +350,16 @@ class Notify:
def __lt__(self, __other): ...
def __ne__(self, __other): ...
class OperationalError(psycopg2.DatabaseError): ...
class ProgrammingError(psycopg2.DatabaseError): ...
class QueryCanceledError(psycopg2.OperationalError): ...
class QuotedString:
adapted: Any
buffer: Any
encoding: Any
def __init__(self, *args, **kwargs) -> None: ...
def getquoted(self, *args, **kwargs): ...
def prepare(self, *args, **kwargs): ...
def __conform__(self, *args, **kwargs): ...
class ReplicationConnection(psycopg2.extensions.connection):
autocommit: Any
isolation_level: Any
replication_type: Any
reset: Any
set_isolation_level: Any
set_session: Any
def __init__(self, *args, **kwargs) -> None: ...
encoding: str
def __init__(self, __str: object, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
@property
def buffer(self) -> Any: ...
def getquoted(self) -> bytes: ...
def prepare(self, __conn: connection) -> None: ...
def __conform__(self, __proto) -> Self | None: ...
class ReplicationCursor(cursor):
feedback_timestamp: Any
@@ -341,9 +380,6 @@ class ReplicationMessage:
wal_end: Any
def __init__(self, *args, **kwargs) -> None: ...
class TransactionRollbackError(psycopg2.OperationalError): ...
class Warning(Exception): ...
class Xid:
bqual: Any
database: Any
@@ -359,21 +395,21 @@ class Xid:
_T_cur = TypeVar("_T_cur", bound=cursor)
class connection:
DataError: Any
DatabaseError: Any
Error: Any
IntegrityError: Any
InterfaceError: Any
InternalError: Any
NotSupportedError: Any
OperationalError: Any
ProgrammingError: Any
Warning: Any
DataError: type[DataError]
DatabaseError: type[DatabaseError]
Error: type[Error]
IntegrityError: type[IntegrityError]
InterfaceError: type[InterfaceError]
InternalError: type[InternalError]
NotSupportedError: type[NotSupportedError]
OperationalError: type[OperationalError]
ProgrammingError: type[ProgrammingError]
Warning: type[Warning]
@property
def async_(self) -> int: ...
autocommit: bool
@property
def binary_types(self) -> Any: ...
def binary_types(self) -> dict[Incomplete, Incomplete]: ...
@property
def closed(self) -> int: ...
cursor_factory: Callable[..., _Cursor]
@@ -387,8 +423,8 @@ class connection:
def isolation_level(self) -> int | None: ...
@isolation_level.setter
def isolation_level(self, __value: str | bytes | int | None) -> None: ...
notices: list[Any]
notifies: list[Any]
notices: list[str]
notifies: list[Notify]
@property
def pgconn_ptr(self) -> int | None: ...
@property
@@ -406,27 +442,33 @@ class connection:
@property
def status(self) -> int: ...
@property
def string_types(self) -> Any: ...
# Really it's dsn: str, async: int = ..., async_: int = ..., but
def string_types(self) -> dict[Incomplete, Incomplete]: ...
# Really it's dsn: str, async: int = 0, async_: int = 0, but
# that would be a syntax error.
def __init__(self, dsn: str, *, async_: int = ...) -> None: ...
def __init__(self, dsn: str, *, async_: int = 0) -> None: ...
def cancel(self) -> None: ...
def close(self) -> None: ...
def commit(self) -> None: ...
@overload
def cursor(self, name: str | bytes | None = ..., *, withhold: bool = ..., scrollable: bool | None = ...) -> _Cursor: ...
def cursor(
self, name: str | bytes | None = None, cursor_factory: None = None, withhold: bool = False, scrollable: bool | None = None
) -> _Cursor: ...
@overload
def cursor(
self,
name: str | bytes | None = ...,
name: str | bytes | None = None,
*,
cursor_factory: Callable[..., _T_cur],
withhold: bool = ...,
scrollable: bool | None = ...,
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
@overload
def cursor(
self, name: str | bytes | None, cursor_factory: Callable[..., _T_cur], withhold: bool = ..., scrollable: bool | None = ...
self,
name: str | bytes | None,
cursor_factory: Callable[..., _T_cur],
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
def fileno(self) -> int: ...
def get_backend_pid(self) -> int: ...
@@ -464,6 +506,17 @@ class connection:
def __enter__(self) -> Self: ...
def __exit__(self, __type: type[BaseException] | None, __name: BaseException | None, __tb: TracebackType | None) -> None: ...
_Connection: TypeAlias = connection
class ReplicationConnection(connection):
autocommit: Any
isolation_level: Any
replication_type: Any
reset: Any
set_isolation_level: Any
set_session: Any
def __init__(self, *args, **kwargs) -> None: ...
class lobject:
closed: Any
mode: Any
@@ -478,24 +531,52 @@ class lobject:
def unlink(self): ...
def write(self, str): ...
def Date(year, month, day): ...
def DateFromPy(*args, **kwargs): ...
def DateFromTicks(ticks): ...
def IntervalFromPy(*args, **kwargs): ...
def Time(hour, minutes, seconds, tzinfo=...): ...
def TimeFromPy(*args, **kwargs): ...
def TimeFromTicks(ticks): ...
def Timestamp(year, month, day, hour, minutes, seconds, tzinfo=...): ...
def TimestampFromPy(*args, **kwargs): ...
def TimestampFromTicks(ticks): ...
@type_check_only
class _datetime:
# The class doesn't exist at runtime but functions below return "psycopg2._psycopg.datetime" objects
# XXX: This and other classes that implement the `ISQLQuote` protocol could be made generic
# in the return type of their `adapted` property if someone asks for it.
def __init__(self, __obj: object, __type: int = -1, **kwargs: Unused) -> None: ...
@property
def adapted(self) -> Any: ...
@property
def type(self) -> int: ...
def getquoted(self) -> bytes: ...
def __conform__(self, __proto) -> Self | None: ...
def Date(__year: int, __month: int, __day: int) -> _datetime: ...
def DateFromPy(__date: dt.date) -> _datetime: ...
def DateFromTicks(__ticks: float) -> _datetime: ...
def IntervalFromPy(__interval: dt.timedelta) -> _datetime: ...
def Time(__hour: int, __minutes: int, __seconds: float, __tzinfo: dt.tzinfo | None = None) -> _datetime: ...
def TimeFromPy(__time: dt.time) -> _datetime: ...
def TimeFromTicks(__ticks: float) -> _datetime: ...
def Timestamp(
__year: int,
__month: int,
__day: int,
__hour: int = 0,
__minutes: int = 0,
__seconds: float = 0,
__tzinfo: dt.tzinfo | None = None,
) -> _datetime: ...
def TimestampFromPy(__datetime: dt.datetime) -> _datetime: ...
def TimestampFromTicks(__ticks: float) -> _datetime: ...
def _connect(*args, **kwargs): ...
def adapt(*args, **kwargs): ...
def encrypt_password(*args, **kwargs): ...
def get_wait_callback(*args, **kwargs): ...
def libpq_version(*args, **kwargs): ...
def new_array_type(oids, name, baseobj): ...
def new_type(oids, name, castobj): ...
def adapt(__obj: object, __protocol: Incomplete = ..., __alternate: Incomplete = ...) -> Any: ...
def encrypt_password(
password: str | bytes, user: str | bytes, scope: connection | cursor | None = None, algorithm: str | None = None
) -> str: ...
def get_wait_callback() -> Incomplete | None: ...
def libpq_version() -> int: ...
def new_array_type(values: tuple[int, ...], name: str, baseobj: _type) -> _type: ...
def new_type(
values: tuple[int, ...],
name: str,
castobj: Callable[[str | bytes | None, cursor], Any] | None = None,
baseobj: Incomplete | None = None,
) -> _type: ...
def parse_dsn(dsn: str | bytes) -> dict[str, Any]: ...
def quote_ident(*args, **kwargs): ...
def register_type(*args, **kwargs): ...
def set_wait_callback(_none): ...
def quote_ident(ident: str | bytes, scope) -> str: ...
def register_type(__obj: _type, __conn_or_curs: connection | cursor | None = None) -> None: ...
def set_wait_callback(__none: Callable[..., Incomplete] | None) -> None: ...

View File

@@ -1,5 +1,8 @@
from _typeshed import Incomplete
from typing import Any
from typing import Any, overload
from typing_extensions import Self
from psycopg2._psycopg import connection, cursor
class Range:
def __init__(
@@ -10,41 +13,50 @@ class Range:
@property
def upper(self): ...
@property
def isempty(self): ...
def isempty(self) -> bool: ...
@property
def lower_inf(self): ...
def lower_inf(self) -> bool: ...
@property
def upper_inf(self): ...
def upper_inf(self) -> bool: ...
@property
def lower_inc(self): ...
def lower_inc(self) -> bool: ...
@property
def upper_inc(self): ...
def __contains__(self, x): ...
def upper_inc(self) -> bool: ...
def __contains__(self, x) -> bool: ...
def __bool__(self) -> bool: ...
def __eq__(self, other): ...
def __ne__(self, other): ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
def __hash__(self) -> int: ...
def __lt__(self, other): ...
def __le__(self, other): ...
def __gt__(self, other): ...
def __ge__(self, other): ...
def __lt__(self, other: Range) -> bool: ...
def __le__(self, other: Range) -> bool: ...
def __gt__(self, other: Range) -> bool: ...
def __ge__(self, other: Range) -> bool: ...
def register_range(pgrange, pyrange, conn_or_curs, globally: bool = False): ...
def register_range(
pgrange: str, pyrange: str | Range, conn_or_curs: connection | cursor, globally: bool = False
) -> RangeCaster: ...
class RangeAdapter:
name: Any
adapted: Any
def __init__(self, adapted) -> None: ...
def __conform__(self, proto): ...
def prepare(self, conn) -> None: ...
def getquoted(self): ...
name: str # this is None here but MUST be str in subclasses
adapted: Range
def __init__(self, adapted: Range) -> None: ...
def __conform__(self, proto) -> Self | None: ...
def prepare(self, conn: connection | None) -> None: ...
def getquoted(self) -> bytes: ...
class RangeCaster:
adapter: type[RangeAdapter]
range: type[Range]
subtype_oid: Any
typecaster: Any
array_typecaster: Any
def __init__(self, pgrange, pyrange, oid, subtype_oid, array_oid: Incomplete | None = None) -> None: ...
def parse(self, s, cur: Incomplete | None = None): ...
def __init__(
self, pgrange: str | RangeAdapter, pyrange: str | Range, oid: int, subtype_oid: int, array_oid: int | None = None
) -> None: ...
@overload
def parse(self, s: None, cur: cursor | None = None) -> None: ...
@overload
def parse(self, s: str, cur: cursor | None = None) -> Range: ...
class NumericRange(Range): ...
class DateRange(Range): ...
@@ -52,11 +64,11 @@ class DateTimeRange(Range): ...
class DateTimeTZRange(Range): ...
class NumberRangeAdapter(RangeAdapter):
def getquoted(self): ...
def getquoted(self) -> bytes: ...
int4range_caster: Any
int8range_caster: Any
numrange_caster: Any
daterange_caster: Any
tsrange_caster: Any
tstzrange_caster: Any
int4range_caster: RangeCaster
int8range_caster: RangeCaster
numrange_caster: RangeCaster
daterange_caster: RangeCaster
tsrange_caster: RangeCaster
tstzrange_caster: RangeCaster

View File

@@ -1,4 +1,4 @@
def lookup(code, _cache={}): ...
def lookup(code: str, _cache: dict[str, str] = {}) -> str: ...
CLASS_SUCCESSFUL_COMPLETION: str
CLASS_WARNING: str
@@ -302,3 +302,5 @@ ASSERT_FAILURE: str
INTERNAL_ERROR: str
DATA_CORRUPTED: str
INDEX_CORRUPTED: str
IDLE_SESSION_TIMEOUT: str
SQL_JSON_ITEM_CANNOT_BE_CAST_TO_TARGET_TYPE: str

View File

@@ -1,11 +1,19 @@
from psycopg2._psycopg import Error as Error, Warning as Warning
from psycopg2._psycopg import (
DatabaseError as DatabaseError,
DataError as DataError,
Error as Error,
IntegrityError as IntegrityError,
InterfaceError as InterfaceError,
InternalError as InternalError,
NotSupportedError as NotSupportedError,
OperationalError as OperationalError,
ProgrammingError as ProgrammingError,
QueryCanceledError as QueryCanceledError,
TransactionRollbackError as TransactionRollbackError,
Warning as Warning,
)
class DatabaseError(Error): ...
class InterfaceError(Error): ...
class DataError(DatabaseError): ...
class DiagnosticsException(DatabaseError): ...
class IntegrityError(DatabaseError): ...
class InternalError(DatabaseError): ...
class InvalidGrantOperation(DatabaseError): ...
class InvalidGrantor(DatabaseError): ...
class InvalidLocatorSpecification(DatabaseError): ...
@@ -14,9 +22,6 @@ class InvalidTransactionInitiation(DatabaseError): ...
class LocatorException(DatabaseError): ...
class NoAdditionalDynamicResultSetsReturned(DatabaseError): ...
class NoData(DatabaseError): ...
class NotSupportedError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...
class SnapshotTooOld(DatabaseError): ...
class SqlStatementNotYetComplete(DatabaseError): ...
class StackedDiagnosticsAccessedWithoutActiveHandler(DatabaseError): ...
@@ -201,7 +206,6 @@ class ProgramLimitExceeded(OperationalError): ...
class ProhibitedSqlStatementAttempted(InternalError): ...
class ProhibitedSqlStatementAttemptedExt(InternalError): ...
class ProtocolViolation(OperationalError): ...
class QueryCanceledError(OperationalError): ...
class RaiseException(InternalError): ...
class ReadOnlySqlTransaction(InternalError): ...
class ReadingSqlDataNotPermitted(InternalError): ...
@@ -235,7 +239,6 @@ class TooManyJsonArrayElements(DataError): ...
class TooManyJsonObjectMembers(DataError): ...
class TooManyRows(InternalError): ...
class TransactionResolutionUnknown(OperationalError): ...
class TransactionRollbackError(OperationalError): ...
class TriggerProtocolViolated(InternalError): ...
class TriggeredDataChangeViolation(OperationalError): ...
class TrimError(DataError): ...
@@ -259,5 +262,7 @@ class SerializationFailure(TransactionRollbackError): ...
class StatementCompletionUnknown(TransactionRollbackError): ...
class TransactionIntegrityConstraintViolation(TransactionRollbackError): ...
class TransactionRollback(TransactionRollbackError): ...
class IdleSessionTimeout(OperationalError): ...
class SqlJsonItemCannotBeCastToTargetType(DataError): ...
def lookup(code): ...
def lookup(code: str) -> type[Error]: ...

View File

@@ -1,5 +1,6 @@
from _typeshed import Incomplete
from typing import Any
from _typeshed import Unused
from collections.abc import Callable, Iterable
from typing import Any, TypeVar, overload
from psycopg2._psycopg import (
BINARYARRAY as BINARYARRAY,
@@ -54,6 +55,8 @@ from psycopg2._psycopg import (
TimestampFromPy as TimestampFromPy,
TransactionRollbackError as TransactionRollbackError,
Xid as Xid,
_ISQLQuoteProto,
_type,
adapt as adapt,
adapters as adapters,
binary_types as binary_types,
@@ -96,20 +99,27 @@ TRANSACTION_STATUS_INTRANS: int
TRANSACTION_STATUS_INERROR: int
TRANSACTION_STATUS_UNKNOWN: int
def register_adapter(typ, callable) -> None: ...
_T = TypeVar("_T")
def register_adapter(typ: type[_T], callable: Callable[[_T], _ISQLQuoteProto]) -> None: ...
class SQL_IN:
def __init__(self, seq) -> None: ...
def prepare(self, conn) -> None: ...
def getquoted(self): ...
def __init__(self, seq: Iterable[object]) -> None: ...
def prepare(self, conn: connection | None) -> None: ...
def getquoted(self) -> bytes: ...
class NoneAdapter:
def __init__(self, obj) -> None: ...
def getquoted(self, _null: bytes = b"NULL"): ...
def __init__(self, obj: Unused) -> None: ...
def getquoted(self, _null: bytes = b"NULL") -> bytes: ...
def make_dsn(dsn: Incomplete | None = None, **kwargs): ...
@overload
def make_dsn(dsn: bytes) -> bytes: ... # type: ignore[misc]
@overload
def make_dsn(dsn: None = None) -> str: ...
@overload
def make_dsn(dsn: str | bytes | None = None, **kwargs: Any) -> str: ...
JSON: Any
JSONARRAY: Any
JSONB: Any
JSONBARRAY: Any
JSON: _type
JSONARRAY: _type | None
JSONB: _type
JSONBARRAY: _type | None

View File

@@ -16,6 +16,9 @@ from psycopg2._psycopg import (
ReplicationConnection as _replicationConnection,
ReplicationCursor as _replicationCursor,
ReplicationMessage as ReplicationMessage,
connection as _connection,
cursor as _cursor,
quote_ident as quote_ident,
)
from psycopg2._range import (
DateRange as DateRange,
@@ -28,8 +31,6 @@ from psycopg2._range import (
register_range as register_range,
)
from .extensions import connection as _connection, cursor as _cursor, quote_ident as quote_ident
_T_cur = TypeVar("_T_cur", bound=_cursor)
class DictCursorBase(_cursor):
@@ -37,19 +38,25 @@ class DictCursorBase(_cursor):
class DictConnection(_connection):
@overload
def cursor(self, name: str | bytes | None = ..., *, withhold: bool = ..., scrollable: bool | None = ...) -> DictCursor: ...
def cursor(
self, name: str | bytes | None = None, cursor_factory: None = None, withhold: bool = False, scrollable: bool | None = None
) -> DictCursor: ...
@overload
def cursor(
self,
name: str | bytes | None = ...,
name: str | bytes | None = None,
*,
cursor_factory: Callable[..., _T_cur],
withhold: bool = ...,
scrollable: bool | None = ...,
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
@overload
def cursor(
self, name: str | bytes | None, cursor_factory: Callable[..., _T_cur], withhold: bool = ..., scrollable: bool | None = ...
self,
name: str | bytes | None,
cursor_factory: Callable[..., _T_cur],
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
class DictCursor(DictCursorBase):
@@ -77,20 +84,24 @@ class DictRow(list[Any]):
class RealDictConnection(_connection):
@overload
def cursor(
self, name: str | bytes | None = ..., *, withhold: bool = ..., scrollable: bool | None = ...
self, name: str | bytes | None = None, cursor_factory: None = None, withhold: bool = False, scrollable: bool | None = None
) -> RealDictCursor: ...
@overload
def cursor(
self,
name: str | bytes | None = ...,
name: str | bytes | None = None,
*,
cursor_factory: Callable[..., _T_cur],
withhold: bool = ...,
scrollable: bool | None = ...,
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
@overload
def cursor(
self, name: str | bytes | None, cursor_factory: Callable[..., _T_cur], withhold: bool = ..., scrollable: bool | None = ...
self,
name: str | bytes | None,
cursor_factory: Callable[..., _T_cur],
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
class RealDictCursor(DictCursorBase):
@@ -110,20 +121,24 @@ class RealDictRow(OrderedDict[Any, Any]):
class NamedTupleConnection(_connection):
@overload
def cursor(
self, name: str | bytes | None = ..., *, withhold: bool = ..., scrollable: bool | None = ...
self, name: str | bytes | None = None, cursor_factory: None = None, withhold: bool = False, scrollable: bool | None = None
) -> NamedTupleCursor: ...
@overload
def cursor(
self,
name: str | bytes | None = ...,
name: str | bytes | None = None,
*,
cursor_factory: Callable[..., _T_cur],
withhold: bool = ...,
scrollable: bool | None = ...,
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
@overload
def cursor(
self, name: str | bytes | None, cursor_factory: Callable[..., _T_cur], withhold: bool = ..., scrollable: bool | None = ...
self,
name: str | bytes | None,
cursor_factory: Callable[..., _T_cur],
withhold: bool = False,
scrollable: bool | None = None,
) -> _T_cur: ...
class NamedTupleCursor(_cursor):

View File

@@ -1,19 +1,20 @@
from _typeshed import Incomplete
from typing import Any
from collections.abc import Hashable
import psycopg2
from psycopg2._psycopg import _AcceptedByInt
class PoolError(psycopg2.Error): ...
class AbstractConnectionPool:
minconn: Any
maxconn: Any
minconn: int
maxconn: int
closed: bool
def __init__(self, minconn, maxconn, *args, **kwargs) -> None: ...
def __init__(self, minconn: _AcceptedByInt, maxconn: _AcceptedByInt, *args, **kwargs) -> None: ...
# getconn, putconn and closeall are officially documented as methods of the
# abstract base class, but in reality, they only exist on the children classes
def getconn(self, key: Incomplete | None = ...): ...
def putconn(self, conn: Any, key: Incomplete | None = ..., close: bool = ...) -> None: ...
def getconn(self, key: Hashable | None = None) -> Incomplete: ...
def putconn(self, conn: Incomplete, key: Hashable | None = None, close: bool = False) -> None: ...
def closeall(self) -> None: ...
class SimpleConnectionPool(AbstractConnectionPool): ...
@@ -21,4 +22,4 @@ class SimpleConnectionPool(AbstractConnectionPool): ...
class ThreadedConnectionPool(AbstractConnectionPool):
# This subclass has a default value for conn which doesn't exist
# in the SimpleConnectionPool class, nor in the documentation
def putconn(self, conn: Incomplete | None = None, key: Incomplete | None = None, close: bool = False) -> None: ...
def putconn(self, conn: Incomplete | None = None, key: Hashable | None = None, close: bool = False) -> None: ...

View File

@@ -1,50 +1,50 @@
from _typeshed import Incomplete
from collections.abc import Iterator
from typing import Any
from _typeshed import SupportsIter
from collections.abc import Iterable, Iterator
from typing import Any, Generic, TypeVar
from psycopg2._psycopg import connection, cursor
_T = TypeVar("_T")
class Composable:
def __init__(self, wrapped) -> None: ...
def as_string(self, context) -> str: ...
def __add__(self, other) -> Composed: ...
def __mul__(self, n) -> Composed: ...
def __eq__(self, other) -> bool: ...
def __ne__(self, other) -> bool: ...
def __init__(self, wrapped: Any) -> None: ...
def as_string(self, context: connection | cursor) -> str: ...
def __add__(self, other: Composable) -> Composed: ...
def __mul__(self, n: int) -> Composed: ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
class Composed(Composable):
def __init__(self, seq) -> None: ...
def __init__(self, seq: Iterable[Composable]) -> None: ...
@property
def seq(self) -> list[Composable]: ...
def as_string(self, context) -> str: ...
def __iter__(self) -> Iterator[Composable]: ...
def __add__(self, other) -> Composed: ...
def join(self, joiner) -> Composed: ...
def __add__(self, other: Composable) -> Composed: ...
def join(self, joiner: str | SQL) -> Composed: ...
class SQL(Composable):
def __init__(self, string) -> None: ...
def __init__(self, string: str) -> None: ...
@property
def string(self) -> str: ...
def as_string(self, context) -> str: ...
def format(self, *args, **kwargs) -> Composed: ...
def join(self, seq) -> Composed: ...
def format(self, *args: Composable, **kwargs: Composable) -> Composed: ...
def join(self, seq: SupportsIter[Composable]) -> Composed: ...
class Identifier(Composable):
def __init__(self, *strings) -> None: ...
def __init__(self, *strings: str) -> None: ...
@property
def strings(self) -> tuple[str, ...]: ...
@property
def string(self) -> str: ...
def as_string(self, context) -> str: ...
class Literal(Composable):
class Literal(Composable, Generic[_T]):
def __init__(self, wrapped: _T) -> None: ...
@property
def wrapped(self): ...
def as_string(self, context) -> str: ...
def wrapped(self) -> _T: ...
class Placeholder(Composable):
def __init__(self, name: Incomplete | None = None) -> None: ...
def __init__(self, name: str | None = None) -> None: ...
@property
def name(self) -> str | None: ...
def as_string(self, context) -> str: ...
NULL: Any
DEFAULT: Any
NULL: SQL
DEFAULT: SQL

View File

@@ -1,26 +1,26 @@
import datetime
from _typeshed import Incomplete
from typing import Any
from typing_extensions import Self
ZERO: Any
ZERO: datetime.timedelta
class FixedOffsetTimezone(datetime.tzinfo):
def __init__(self, offset: Incomplete | None = None, name: Incomplete | None = None) -> None: ...
def __new__(cls, offset: Incomplete | None = None, name: Incomplete | None = None): ...
def __eq__(self, other): ...
def __ne__(self, other): ...
def __getinitargs__(self): ...
def utcoffset(self, dt): ...
def tzname(self, dt): ...
def dst(self, dt): ...
def __init__(self, offset: datetime.timedelta | float | None = None, name: str | None = None) -> None: ...
def __new__(cls, offset: datetime.timedelta | float | None = None, name: str | None = None) -> Self: ...
def __eq__(self, other: object) -> bool: ...
def __ne__(self, other: object) -> bool: ...
def __getinitargs__(self) -> tuple[Any, ...]: ...
def utcoffset(self, dt: datetime.datetime | None) -> datetime.timedelta: ...
def tzname(self, dt: datetime.datetime | None) -> str: ...
def dst(self, dt: datetime.datetime | None) -> datetime.timedelta: ...
STDOFFSET: Any
DSTOFFSET: Any
DSTDIFF: Any
STDOFFSET: datetime.timedelta
DSTOFFSET: datetime.timedelta
DSTDIFF: datetime.timedelta
class LocalTimezone(datetime.tzinfo):
def utcoffset(self, dt): ...
def dst(self, dt): ...
def tzname(self, dt): ...
def utcoffset(self, dt: datetime.datetime) -> datetime.timedelta: ... # type: ignore[override]
def dst(self, dt: datetime.datetime) -> datetime.timedelta: ... # type: ignore[override]
def tzname(self, dt: datetime.datetime) -> str: ... # type: ignore[override]
LOCAL: Any
LOCAL: LocalTimezone