Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/dialects/postgresql»

Browse Source
This commit is contained in:
Polina
2026-07-02 20:23:46 +00:00
parent 6392e58473
commit 8eb4fab879
5 changed files with 2151 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
venv/Lib/site-packages/sqlalchemy/dialects/postgresql/operators.py Normal file
View File

@@ -0,0 +1,128 @@
# dialects/postgresql/operators.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from ...sql import operators
_getitem_precedence = operators._PRECEDENCE[operators.json_getitem_op]
_eq_precedence = operators._PRECEDENCE[operators.eq]
# JSON + JSONB
ASTEXT = operators.custom_op(
"->>",
precedence=_getitem_precedence,
natural_self_precedent=True,
eager_grouping=True,
)
JSONPATH_ASTEXT = operators.custom_op(
"#>>",
precedence=_getitem_precedence,
natural_self_precedent=True,
eager_grouping=True,
)
# JSONB + HSTORE
HAS_KEY = operators.custom_op(
"?",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
HAS_ALL = operators.custom_op(
"?&",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
HAS_ANY = operators.custom_op(
"?|",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
# JSONB
DELETE_PATH = operators.custom_op(
"#-",
precedence=_getitem_precedence,
natural_self_precedent=True,
eager_grouping=True,
)
PATH_EXISTS = operators.custom_op(
"@?",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
PATH_MATCH = operators.custom_op(
"@@",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
# JSONB + ARRAY + HSTORE + RANGE
CONTAINS = operators.custom_op(
"@>",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
CONTAINED_BY = operators.custom_op(
"<@",
precedence=_eq_precedence,
natural_self_precedent=True,
eager_grouping=True,
is_comparison=True,
)
# ARRAY + RANGE
OVERLAP = operators.custom_op(
"&&",
precedence=_eq_precedence,
is_comparison=True,
)
# RANGE
STRICTLY_LEFT_OF = operators.custom_op(
"<<", precedence=_eq_precedence, is_comparison=True
)
STRICTLY_RIGHT_OF = operators.custom_op(
">>", precedence=_eq_precedence, is_comparison=True
)
NOT_EXTEND_RIGHT_OF = operators.custom_op(
"&<", precedence=_eq_precedence, is_comparison=True
)
NOT_EXTEND_LEFT_OF = operators.custom_op(
"&>", precedence=_eq_precedence, is_comparison=True
)
ADJACENT_TO = operators.custom_op(
"-|-", precedence=_eq_precedence, is_comparison=True
)
# HSTORE
GETITEM = operators.custom_op(
"->",
precedence=_getitem_precedence,
natural_self_precedent=True,
eager_grouping=True,
)

667
venv/Lib/site-packages/sqlalchemy/dialects/postgresql/pg8000.py Normal file
View File

@@ -0,0 +1,667 @@
# dialects/postgresql/pg8000.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors <see AUTHORS
# file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
r"""
.. dialect:: postgresql+pg8000
:name: pg8000
:dbapi: pg8000
:connectstring: postgresql+pg8000://user:password@host:port/dbname[?key=value&key=value...]
:url: https://pypi.org/project/pg8000/
.. versionchanged:: 1.4 The pg8000 dialect has been updated for version
1.16.6 and higher, and is again part of SQLAlchemy's continuous integration
with full feature support.
.. _pg8000_unicode:
Unicode
-------
pg8000 will encode / decode string values between it and the server using the
PostgreSQL ``client_encoding`` parameter; by default this is the value in
the ``postgresql.conf`` file, which often defaults to ``SQL_ASCII``.
Typically, this can be changed to ``utf-8``, as a more useful default::
# client_encoding = sql_ascii # actually, defaults to database encoding
client_encoding = utf8
The ``client_encoding`` can be overridden for a session by executing the SQL:
.. sourcecode:: sql
SET CLIENT_ENCODING TO 'utf8';
SQLAlchemy will execute this SQL on all new connections based on the value
passed to :func:`_sa.create_engine` using the ``client_encoding`` parameter::
engine = create_engine(
"postgresql+pg8000://user:pass@host/dbname", client_encoding="utf8"
)
.. _pg8000_ssl:
SSL Connections
---------------
pg8000 accepts a Python ``SSLContext`` object which may be specified using the
:paramref:`_sa.create_engine.connect_args` dictionary::
import ssl
ssl_context = ssl.create_default_context()
engine = sa.create_engine(
"postgresql+pg8000://scott:tiger@192.168.0.199/test",
connect_args={"ssl_context": ssl_context},
)
If the server uses an automatically-generated certificate that is self-signed
or does not match the host name (as seen from the client), it may also be
necessary to disable hostname checking::
import ssl
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
engine = sa.create_engine(
"postgresql+pg8000://scott:tiger@192.168.0.199/test",
connect_args={"ssl_context": ssl_context},
)
.. _pg8000_isolation_level:
pg8000 Transaction Isolation Level
-------------------------------------
The pg8000 dialect offers the same isolation level settings as that
of the :ref:`psycopg2 <psycopg2_isolation_level>` dialect:
* ``READ COMMITTED``
* ``READ UNCOMMITTED``
* ``REPEATABLE READ``
* ``SERIALIZABLE``
* ``AUTOCOMMIT``
.. seealso::
:ref:`postgresql_isolation_level`
:ref:`psycopg2_isolation_level`
""" # noqa
import decimal
import re
from . import ranges
from .array import ARRAY as PGARRAY
from .base import _DECIMAL_TYPES
from .base import _FLOAT_TYPES
from .base import _INT_TYPES
from .base import ENUM
from .base import INTERVAL
from .base import PGCompiler
from .base import PGDialect
from .base import PGExecutionContext
from .base import PGIdentifierPreparer
from .json import JSON
from .json import JSONB
from .json import JSONPathType
from .pg_catalog import _SpaceVector
from .pg_catalog import OIDVECTOR
from .types import CITEXT
from ... import exc
from ... import util
from ...engine import processors
from ...sql import sqltypes
from ...sql.elements import quoted_name
class _PGString(sqltypes.String):
render_bind_cast = True
class _PGNumeric(sqltypes.Numeric):
render_bind_cast = True
def result_processor(self, dialect, coltype):
if self.asdecimal:
if coltype in _FLOAT_TYPES:
return processors.to_decimal_processor_factory(
decimal.Decimal, self._effective_decimal_return_scale
)
elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
# pg8000 returns Decimal natively for 1700
return None
else:
raise exc.InvalidRequestError(
"Unknown PG numeric type: %d" % coltype
)
else:
if coltype in _FLOAT_TYPES:
# pg8000 returns float natively for 701
return None
elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
return processors.to_float
else:
raise exc.InvalidRequestError(
"Unknown PG numeric type: %d" % coltype
)
class _PGFloat(_PGNumeric, sqltypes.Float):
__visit_name__ = "float"
render_bind_cast = True
class _PGNumericNoBind(_PGNumeric):
def bind_processor(self, dialect):
return None
class _PGJSON(JSON):
render_bind_cast = True
def result_processor(self, dialect, coltype):
return None
class _PGJSONB(JSONB):
render_bind_cast = True
def result_processor(self, dialect, coltype):
return None
class _PGJSONIndexType(sqltypes.JSON.JSONIndexType):
def get_dbapi_type(self, dbapi):
raise NotImplementedError("should not be here")
class _PGJSONIntIndexType(sqltypes.JSON.JSONIntIndexType):
__visit_name__ = "json_int_index"
render_bind_cast = True
class _PGJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
__visit_name__ = "json_str_index"
render_bind_cast = True
class _PGJSONPathType(JSONPathType):
pass
# DBAPI type 1009
class _PGEnum(ENUM):
def get_dbapi_type(self, dbapi):
return dbapi.UNKNOWN
class _PGInterval(INTERVAL):
render_bind_cast = True
def get_dbapi_type(self, dbapi):
return dbapi.INTERVAL
@classmethod
def adapt_emulated_to_native(cls, interval, **kw):
return _PGInterval(precision=interval.second_precision)
class _PGTimeStamp(sqltypes.DateTime):
render_bind_cast = True
class _PGDate(sqltypes.Date):
render_bind_cast = True
class _PGTime(sqltypes.Time):
render_bind_cast = True
class _PGInteger(sqltypes.Integer):
render_bind_cast = True
class _PGSmallInteger(sqltypes.SmallInteger):
render_bind_cast = True
class _PGNullType(sqltypes.NullType):
pass
class _PGBigInteger(sqltypes.BigInteger):
render_bind_cast = True
class _PGBoolean(sqltypes.Boolean):
render_bind_cast = True
class _PGARRAY(PGARRAY):
render_bind_cast = True
class _PGOIDVECTOR(_SpaceVector, OIDVECTOR):
pass
class _Pg8000Range(ranges.AbstractSingleRangeImpl):
def bind_processor(self, dialect):
pg8000_Range = dialect.dbapi.Range
def to_range(value):
if isinstance(value, ranges.Range):
value = pg8000_Range(
value.lower, value.upper, value.bounds, value.empty
)
return value
return to_range
def result_processor(self, dialect, coltype):
def to_range(value):
if value is not None:
value = ranges.Range(
value.lower,
value.upper,
bounds=value.bounds,
empty=value.is_empty,
)
return value
return to_range
class _Pg8000MultiRange(ranges.AbstractMultiRangeImpl):
def bind_processor(self, dialect):
pg8000_Range = dialect.dbapi.Range
def to_multirange(value):
if isinstance(value, list):
mr = []
for v in value:
if isinstance(v, ranges.Range):
mr.append(
pg8000_Range(v.lower, v.upper, v.bounds, v.empty)
)
else:
mr.append(v)
return mr
else:
return value
return to_multirange
def result_processor(self, dialect, coltype):
def to_multirange(value):
if value is None:
return None
else:
return ranges.MultiRange(
ranges.Range(
v.lower, v.upper, bounds=v.bounds, empty=v.is_empty
)
for v in value
)
return to_multirange
_server_side_id = util.counter()
class PGExecutionContext_pg8000(PGExecutionContext):
def create_server_side_cursor(self):
ident = "c_%s_%s" % (hex(id(self))[2:], hex(_server_side_id())[2:])
return ServerSideCursor(self._dbapi_connection.cursor(), ident)
def pre_exec(self):
if not self.compiled:
return
class ServerSideCursor:
server_side = True
def __init__(self, cursor, ident):
self.ident = ident
self.cursor = cursor
@property
def connection(self):
return self.cursor.connection
@property
def rowcount(self):
return self.cursor.rowcount
@property
def description(self):
return self.cursor.description
def execute(self, operation, args=(), stream=None):
op = "DECLARE " + self.ident + " NO SCROLL CURSOR FOR " + operation
self.cursor.execute(op, args, stream=stream)
return self
def executemany(self, operation, param_sets):
self.cursor.executemany(operation, param_sets)
return self
def fetchone(self):
self.cursor.execute("FETCH FORWARD 1 FROM " + self.ident)
return self.cursor.fetchone()
def fetchmany(self, num=None):
if num is None:
return self.fetchall()
else:
self.cursor.execute(
"FETCH FORWARD " + str(int(num)) + " FROM " + self.ident
)
return self.cursor.fetchall()
def fetchall(self):
self.cursor.execute("FETCH FORWARD ALL FROM " + self.ident)
return self.cursor.fetchall()
def close(self):
self.cursor.execute("CLOSE " + self.ident)
self.cursor.close()
def setinputsizes(self, *sizes):
self.cursor.setinputsizes(*sizes)
def setoutputsize(self, size, column=None):
pass
class PGCompiler_pg8000(PGCompiler):
def visit_mod_binary(self, binary, operator, **kw):
return (
self.process(binary.left, **kw)
+ " %% "
+ self.process(binary.right, **kw)
)
class PGIdentifierPreparer_pg8000(PGIdentifierPreparer):
def __init__(self, *args, **kwargs):
PGIdentifierPreparer.__init__(self, *args, **kwargs)
self._double_percents = False
class PGDialect_pg8000(PGDialect):
driver = "pg8000"
supports_statement_cache = True
supports_unicode_statements = True
supports_unicode_binds = True
default_paramstyle = "format"
supports_sane_multi_rowcount = True
execution_ctx_cls = PGExecutionContext_pg8000
statement_compiler = PGCompiler_pg8000
preparer = PGIdentifierPreparer_pg8000
supports_server_side_cursors = True
render_bind_cast = True
# reversed as of pg8000 1.16.6. 1.16.5 and lower
# are no longer compatible
description_encoding = None
# description_encoding = "use_encoding"
colspecs = util.update_copy(
PGDialect.colspecs,
{
sqltypes.String: _PGString,
sqltypes.Numeric: _PGNumericNoBind,
sqltypes.Float: _PGFloat,
sqltypes.JSON: _PGJSON,
sqltypes.Boolean: _PGBoolean,
sqltypes.NullType: _PGNullType,
JSONB: _PGJSONB,
CITEXT: CITEXT,
sqltypes.JSON.JSONPathType: _PGJSONPathType,
sqltypes.JSON.JSONIndexType: _PGJSONIndexType,
sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
sqltypes.Interval: _PGInterval,
INTERVAL: _PGInterval,
sqltypes.DateTime: _PGTimeStamp,
sqltypes.DateTime: _PGTimeStamp,
sqltypes.Date: _PGDate,
sqltypes.Time: _PGTime,
sqltypes.Integer: _PGInteger,
sqltypes.SmallInteger: _PGSmallInteger,
sqltypes.BigInteger: _PGBigInteger,
sqltypes.Enum: _PGEnum,
sqltypes.ARRAY: _PGARRAY,
OIDVECTOR: _PGOIDVECTOR,
ranges.INT4RANGE: _Pg8000Range,
ranges.INT8RANGE: _Pg8000Range,
ranges.NUMRANGE: _Pg8000Range,
ranges.DATERANGE: _Pg8000Range,
ranges.TSRANGE: _Pg8000Range,
ranges.TSTZRANGE: _Pg8000Range,
ranges.INT4MULTIRANGE: _Pg8000MultiRange,
ranges.INT8MULTIRANGE: _Pg8000MultiRange,
ranges.NUMMULTIRANGE: _Pg8000MultiRange,
ranges.DATEMULTIRANGE: _Pg8000MultiRange,
ranges.TSMULTIRANGE: _Pg8000MultiRange,
ranges.TSTZMULTIRANGE: _Pg8000MultiRange,
},
)
def __init__(self, client_encoding=None, **kwargs):
PGDialect.__init__(self, **kwargs)
self.client_encoding = client_encoding
if self._dbapi_version < (1, 16, 6):
raise NotImplementedError("pg8000 1.16.6 or greater is required")
if self._native_inet_types:
raise NotImplementedError(
"The pg8000 dialect does not fully implement "
"ipaddress type handling; INET is supported by default, "
"CIDR is not"
)
@util.memoized_property
def _dbapi_version(self):
if self.dbapi and hasattr(self.dbapi, "__version__"):
return tuple(
[
int(x)
for x in re.findall(
r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
)
]
)
else:
return (99, 99, 99)
@classmethod
def import_dbapi(cls):
return __import__("pg8000")
def create_connect_args(self, url):
opts = url.translate_connect_args(username="user")
if "port" in opts:
opts["port"] = int(opts["port"])
opts.update(url.query)
return ([], opts)
def is_disconnect(self, e, connection, cursor):
if isinstance(e, self.dbapi.InterfaceError) and "network error" in str(
e
):
# new as of pg8000 1.19.0 for broken connections
return True
# connection was closed normally
return "connection is closed" in str(e)
def get_isolation_level_values(self, dbapi_connection):
return (
"AUTOCOMMIT",
"READ COMMITTED",
"READ UNCOMMITTED",
"REPEATABLE READ",
"SERIALIZABLE",
)
def set_isolation_level(self, dbapi_connection, level):
level = level.replace("_", " ")
if level == "AUTOCOMMIT":
dbapi_connection.autocommit = True
else:
dbapi_connection.autocommit = False
cursor = dbapi_connection.cursor()
cursor.execute(
"SET SESSION CHARACTERISTICS AS TRANSACTION "
f"ISOLATION LEVEL {level}"
)
cursor.execute("COMMIT")
cursor.close()
def detect_autocommit_setting(self, dbapi_conn) -> bool:
return bool(dbapi_conn.autocommit)
def set_readonly(self, connection, value):
cursor = connection.cursor()
try:
cursor.execute(
"SET SESSION CHARACTERISTICS AS TRANSACTION %s"
% ("READ ONLY" if value else "READ WRITE")
)
cursor.execute("COMMIT")
finally:
cursor.close()
def get_readonly(self, connection):
cursor = connection.cursor()
try:
cursor.execute("show transaction_read_only")
val = cursor.fetchone()[0]
finally:
cursor.close()
return val == "on"
def set_deferrable(self, connection, value):
cursor = connection.cursor()
try:
cursor.execute(
"SET SESSION CHARACTERISTICS AS TRANSACTION %s"
% ("DEFERRABLE" if value else "NOT DEFERRABLE")
)
cursor.execute("COMMIT")
finally:
cursor.close()
def get_deferrable(self, connection):
cursor = connection.cursor()
try:
cursor.execute("show transaction_deferrable")
val = cursor.fetchone()[0]
finally:
cursor.close()
return val == "on"
def _set_client_encoding(self, dbapi_connection, client_encoding):
cursor = dbapi_connection.cursor()
cursor.execute(f"""
SET CLIENT_ENCODING TO '{client_encoding.replace("'", "''")}'""")
cursor.execute("COMMIT")
cursor.close()
def do_begin_twophase(self, connection, xid):
connection.connection.tpc_begin((0, xid, ""))
def do_prepare_twophase(self, connection, xid):
connection.connection.tpc_prepare()
def do_rollback_twophase(
self, connection, xid, is_prepared=True, recover=False
):
connection.connection.tpc_rollback((0, xid, ""))
def do_commit_twophase(
self, connection, xid, is_prepared=True, recover=False
):
connection.connection.tpc_commit((0, xid, ""))
def do_recover_twophase(self, connection):
return [row[1] for row in connection.connection.tpc_recover()]
def on_connect(self):
fns = []
def on_connect(conn):
conn.py_types[quoted_name] = conn.py_types[str]
fns.append(on_connect)
if self.client_encoding is not None:
def on_connect(conn):
self._set_client_encoding(conn, self.client_encoding)
fns.append(on_connect)
if self._native_inet_types is False:
def on_connect(conn):
# inet
conn.register_in_adapter(869, lambda s: s)
# cidr
conn.register_in_adapter(650, lambda s: s)
fns.append(on_connect)
if self._json_deserializer:
def on_connect(conn):
# json
conn.register_in_adapter(114, self._json_deserializer)
# jsonb
conn.register_in_adapter(3802, self._json_deserializer)
fns.append(on_connect)
if len(fns) > 0:
def on_connect(conn):
for fn in fns:
fn(conn)
return on_connect
else:
return None
@util.memoized_property
def _dialect_specific_select_one(self):
return ";"
dialect = PGDialect_pg8000

326
venv/Lib/site-packages/sqlalchemy/dialects/postgresql/pg_catalog.py Normal file
View File

@@ -0,0 +1,326 @@
# dialects/postgresql/pg_catalog.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
from typing import Any
from typing import Optional
from typing import Sequence
from typing import TYPE_CHECKING
from .array import ARRAY
from .types import OID
from .types import REGCLASS
from ... import Column
from ... import func
from ... import MetaData
from ... import Table
from ...types import BigInteger
from ...types import Boolean
from ...types import CHAR
from ...types import Float
from ...types import Integer
from ...types import SmallInteger
from ...types import String
from ...types import Text
from ...types import TypeDecorator
if TYPE_CHECKING:
from ...engine.interfaces import Dialect
from ...sql.type_api import _ResultProcessorType
# types
class NAME(TypeDecorator[str]):
impl = String(64, collation="C")
cache_ok = True
class PG_NODE_TREE(TypeDecorator[str]):
impl = Text(collation="C")
cache_ok = True
class INT2VECTOR(TypeDecorator[Sequence[int]]):
impl = ARRAY(SmallInteger)
cache_ok = True
class OIDVECTOR(TypeDecorator[Sequence[int]]):
impl = ARRAY(OID)
cache_ok = True
class _SpaceVector:
def result_processor(
self, dialect: Dialect, coltype: object
) -> _ResultProcessorType[list[int]]:
def process(value: Any) -> Optional[list[int]]:
if value is None:
return value
return [int(p) for p in value.split(" ")]
return process
REGPROC = REGCLASS # seems an alias
# functions
_pg_cat = func.pg_catalog
quote_ident = _pg_cat.quote_ident
pg_table_is_visible = _pg_cat.pg_table_is_visible
pg_type_is_visible = _pg_cat.pg_type_is_visible
pg_get_viewdef = _pg_cat.pg_get_viewdef
pg_get_serial_sequence = _pg_cat.pg_get_serial_sequence
format_type = _pg_cat.format_type
pg_get_expr = _pg_cat.pg_get_expr
pg_get_constraintdef = _pg_cat.pg_get_constraintdef
pg_get_indexdef = _pg_cat.pg_get_indexdef
# constants
RELKINDS_TABLE_NO_FOREIGN = ("r", "p")
RELKINDS_TABLE = RELKINDS_TABLE_NO_FOREIGN + ("f",)
RELKINDS_VIEW = ("v",)
RELKINDS_MAT_VIEW = ("m",)
RELKINDS_ALL_TABLE_LIKE = RELKINDS_TABLE + RELKINDS_VIEW + RELKINDS_MAT_VIEW
# tables
pg_catalog_meta = MetaData(schema="pg_catalog")
pg_namespace = Table(
"pg_namespace",
pg_catalog_meta,
Column("oid", OID),
Column("nspname", NAME),
Column("nspowner", OID),
)
pg_class = Table(
"pg_class",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("relname", NAME),
Column("relnamespace", OID),
Column("reltype", OID),
Column("reloftype", OID),
Column("relowner", OID),
Column("relam", OID),
Column("relfilenode", OID),
Column("reltablespace", OID),
Column("relpages", Integer),
Column("reltuples", Float),
Column("relallvisible", Integer, info={"server_version": (9, 2)}),
Column("reltoastrelid", OID),
Column("relhasindex", Boolean),
Column("relisshared", Boolean),
Column("relpersistence", CHAR, info={"server_version": (9, 1)}),
Column("relkind", CHAR),
Column("relnatts", SmallInteger),
Column("relchecks", SmallInteger),
Column("relhasrules", Boolean),
Column("relhastriggers", Boolean),
Column("relhassubclass", Boolean),
Column("relrowsecurity", Boolean),
Column("relforcerowsecurity", Boolean, info={"server_version": (9, 5)}),
Column("relispopulated", Boolean, info={"server_version": (9, 3)}),
Column("relreplident", CHAR, info={"server_version": (9, 4)}),
Column("relispartition", Boolean, info={"server_version": (10,)}),
Column("relrewrite", OID, info={"server_version": (11,)}),
Column("reloptions", ARRAY(Text)),
)
pg_type = Table(
"pg_type",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("typname", NAME),
Column("typnamespace", OID),
Column("typowner", OID),
Column("typlen", SmallInteger),
Column("typbyval", Boolean),
Column("typtype", CHAR),
Column("typcategory", CHAR),
Column("typispreferred", Boolean),
Column("typisdefined", Boolean),
Column("typdelim", CHAR),
Column("typrelid", OID),
Column("typelem", OID),
Column("typarray", OID),
Column("typinput", REGPROC),
Column("typoutput", REGPROC),
Column("typreceive", REGPROC),
Column("typsend", REGPROC),
Column("typmodin", REGPROC),
Column("typmodout", REGPROC),
Column("typanalyze", REGPROC),
Column("typalign", CHAR),
Column("typstorage", CHAR),
Column("typnotnull", Boolean),
Column("typbasetype", OID),
Column("typtypmod", Integer),
Column("typndims", Integer),
Column("typcollation", OID, info={"server_version": (9, 1)}),
Column("typdefault", Text),
)
pg_index = Table(
"pg_index",
pg_catalog_meta,
Column("indexrelid", OID),
Column("indrelid", OID),
Column("indnatts", SmallInteger),
Column("indnkeyatts", SmallInteger, info={"server_version": (11,)}),
Column("indisunique", Boolean),
Column("indnullsnotdistinct", Boolean, info={"server_version": (15,)}),
Column("indisprimary", Boolean),
Column("indisexclusion", Boolean, info={"server_version": (9, 1)}),
Column("indimmediate", Boolean),
Column("indisclustered", Boolean),
Column("indisvalid", Boolean),
Column("indcheckxmin", Boolean),
Column("indisready", Boolean),
Column("indislive", Boolean, info={"server_version": (9, 3)}), # 9.3
Column("indisreplident", Boolean),
Column("indkey", INT2VECTOR),
Column("indcollation", OIDVECTOR, info={"server_version": (9, 1)}), # 9.1
Column("indclass", OIDVECTOR),
Column("indoption", INT2VECTOR),
Column("indexprs", PG_NODE_TREE),
Column("indpred", PG_NODE_TREE),
)
pg_attribute = Table(
"pg_attribute",
pg_catalog_meta,
Column("attrelid", OID),
Column("attname", NAME),
Column("atttypid", OID),
Column("attstattarget", Integer),
Column("attlen", SmallInteger),
Column("attnum", SmallInteger),
Column("attndims", Integer),
Column("attcacheoff", Integer),
Column("atttypmod", Integer),
Column("attbyval", Boolean),
Column("attstorage", CHAR),
Column("attalign", CHAR),
Column("attnotnull", Boolean),
Column("atthasdef", Boolean),
Column("atthasmissing", Boolean, info={"server_version": (11,)}),
Column("attidentity", CHAR, info={"server_version": (10,)}),
Column("attgenerated", CHAR, info={"server_version": (12,)}),
Column("attisdropped", Boolean),
Column("attislocal", Boolean),
Column("attinhcount", Integer),
Column("attcollation", OID, info={"server_version": (9, 1)}),
)
pg_constraint = Table(
"pg_constraint",
pg_catalog_meta,
Column("oid", OID), # 9.3
Column("conname", NAME),
Column("connamespace", OID),
Column("contype", CHAR),
Column("condeferrable", Boolean),
Column("condeferred", Boolean),
Column("convalidated", Boolean, info={"server_version": (9, 1)}),
Column("conrelid", OID),
Column("contypid", OID),
Column("conindid", OID),
Column("conparentid", OID, info={"server_version": (11,)}),
Column("confrelid", OID),
Column("confupdtype", CHAR),
Column("confdeltype", CHAR),
Column("confmatchtype", CHAR),
Column("conislocal", Boolean),
Column("coninhcount", Integer),
Column("connoinherit", Boolean, info={"server_version": (9, 2)}),
Column("conkey", ARRAY(SmallInteger)),
Column("confkey", ARRAY(SmallInteger)),
)
pg_sequence = Table(
"pg_sequence",
pg_catalog_meta,
Column("seqrelid", OID),
Column("seqtypid", OID),
Column("seqstart", BigInteger),
Column("seqincrement", BigInteger),
Column("seqmax", BigInteger),
Column("seqmin", BigInteger),
Column("seqcache", BigInteger),
Column("seqcycle", Boolean),
info={"server_version": (10,)},
)
pg_attrdef = Table(
"pg_attrdef",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("adrelid", OID),
Column("adnum", SmallInteger),
Column("adbin", PG_NODE_TREE),
)
pg_description = Table(
"pg_description",
pg_catalog_meta,
Column("objoid", OID),
Column("classoid", OID),
Column("objsubid", Integer),
Column("description", Text(collation="C")),
)
pg_enum = Table(
"pg_enum",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("enumtypid", OID),
Column("enumsortorder", Float(), info={"server_version": (9, 1)}),
Column("enumlabel", NAME),
)
pg_am = Table(
"pg_am",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("amname", NAME),
Column("amhandler", REGPROC, info={"server_version": (9, 6)}),
Column("amtype", CHAR, info={"server_version": (9, 6)}),
)
pg_collation = Table(
"pg_collation",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("collname", NAME),
Column("collnamespace", OID),
Column("collowner", OID),
Column("collprovider", CHAR, info={"server_version": (10,)}),
Column("collisdeterministic", Boolean, info={"server_version": (12,)}),
Column("collencoding", Integer),
Column("collcollate", Text),
Column("collctype", Text),
Column("colliculocale", Text),
Column("collicurules", Text, info={"server_version": (16,)}),
Column("collversion", Text, info={"server_version": (10,)}),
)
pg_opclass = Table(
"pg_opclass",
pg_catalog_meta,
Column("oid", OID, info={"server_version": (9, 3)}),
Column("opcmethod", NAME),
Column("opcname", NAME),
Column("opsnamespace", OID),
Column("opsowner", OID),
Column("opcfamily", OID),
Column("opcintype", OID),
Column("opcdefault", Boolean),
Column("opckeytype", OID),
)

184
venv/Lib/site-packages/sqlalchemy/dialects/postgresql/provision.py Normal file
View File

@@ -0,0 +1,184 @@
# dialects/postgresql/provision.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
import time
from ... import exc
from ... import inspect
from ... import text
from ...testing import warn_test_suite
from ...testing.provision import create_db
from ...testing.provision import drop_all_schema_objects_post_tables
from ...testing.provision import drop_all_schema_objects_pre_tables
from ...testing.provision import drop_db
from ...testing.provision import log
from ...testing.provision import post_configure_engine
from ...testing.provision import prepare_for_drop_tables
from ...testing.provision import set_default_schema_on_connection
from ...testing.provision import temp_table_keyword_args
from ...testing.provision import upsert
@create_db.for_db("postgresql")
def _pg_create_db(cfg, eng, ident):
template_db = cfg.options.postgresql_templatedb
with eng.execution_options(isolation_level="AUTOCOMMIT").begin() as conn:
if not template_db:
template_db = conn.exec_driver_sql(
"select current_database()"
).scalar()
attempt = 0
while True:
try:
conn.exec_driver_sql(
"CREATE DATABASE %s TEMPLATE %s" % (ident, template_db)
)
except exc.OperationalError as err:
attempt += 1
if attempt >= 3:
raise
if "accessed by other users" in str(err):
log.info(
"Waiting to create %s, URI %r, "
"template DB %s is in use sleeping for .5",
ident,
eng.url,
template_db,
)
time.sleep(0.5)
except:
raise
else:
break
@drop_db.for_db("postgresql")
def _pg_drop_db(cfg, eng, ident):
with eng.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
with conn.begin():
conn.execute(
text(
"select pg_terminate_backend(pid) from pg_stat_activity "
"where usename=current_user and pid != pg_backend_pid() "
"and datname=:dname"
),
dict(dname=ident),
)
conn.exec_driver_sql("DROP DATABASE %s" % ident)
@temp_table_keyword_args.for_db("postgresql")
def _postgresql_temp_table_keyword_args(cfg, eng):
return {"prefixes": ["TEMPORARY"]}
@set_default_schema_on_connection.for_db("postgresql")
def _postgresql_set_default_schema_on_connection(
cfg, dbapi_connection, schema_name
):
existing_autocommit = dbapi_connection.autocommit
dbapi_connection.autocommit = True
cursor = dbapi_connection.cursor()
cursor.execute("SET SESSION search_path='%s'" % schema_name)
cursor.close()
dbapi_connection.autocommit = existing_autocommit
@drop_all_schema_objects_pre_tables.for_db("postgresql")
def drop_all_schema_objects_pre_tables(cfg, eng):
with eng.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
for xid in conn.exec_driver_sql(
"SELECT gid FROM pg_prepared_xacts "
"WHERE database = current_database()"
).scalars():
eng.dialect.do_rollback_twophase(conn, xid, recover=True)
@drop_all_schema_objects_post_tables.for_db("postgresql")
def drop_all_schema_objects_post_tables(cfg, eng):
from sqlalchemy.dialects import postgresql
inspector = inspect(eng)
with eng.begin() as conn:
for enum in inspector.get_enums("*"):
conn.execute(
postgresql.DropEnumType(
postgresql.ENUM(name=enum["name"], schema=enum["schema"])
)
)
@prepare_for_drop_tables.for_db("postgresql")
def prepare_for_drop_tables(config, connection):
"""Ensure there are no locks on the current username/database."""
result = connection.exec_driver_sql(
"select pid, state, wait_event_type, query "
# "select pg_terminate_backend(pid), state, wait_event_type "
"from pg_stat_activity where "
"usename=current_user "
"and datname=current_database() and state='idle in transaction' "
"and pid != pg_backend_pid()"
)
rows = result.all() # noqa
if rows:
warn_test_suite(
"PostgreSQL may not be able to DROP tables due to "
"idle in transaction: %s"
% ("; ".join(row._mapping["query"] for row in rows))
)
@upsert.for_db("postgresql")
def _upsert(
cfg,
table,
returning,
*,
set_lambda=None,
sort_by_parameter_order=False,
index_elements=None,
):
from sqlalchemy.dialects.postgresql import insert
stmt = insert(table)
table_pk = inspect(table).selectable
if set_lambda:
if index_elements is None:
index_elements = table_pk.primary_key
stmt = stmt.on_conflict_do_update(
index_elements=index_elements, set_=set_lambda(stmt.excluded)
)
else:
stmt = stmt.on_conflict_do_nothing()
stmt = stmt.returning(
*returning, sort_by_parameter_order=sort_by_parameter_order
)
return stmt
_extensions = [
("citext", (13,)),
("hstore", (13,)),
]
@post_configure_engine.for_db("postgresql")
def _create_citext_extension(url, engine, follower_ident):
with engine.connect() as conn:
for extension, min_version in _extensions:
if conn.dialect.server_version_info >= min_version:
conn.execute(
text(f"CREATE EXTENSION IF NOT EXISTS {extension}")
)
conn.commit()

846
venv/Lib/site-packages/sqlalchemy/dialects/postgresql/psycopg.py Normal file
View File

@@ -0,0 +1,846 @@
# dialects/postgresql/psycopg.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
r"""
.. dialect:: postgresql+psycopg
:name: psycopg (a.k.a. psycopg 3)
:dbapi: psycopg
:connectstring: postgresql+psycopg://user:password@host:port/dbname[?key=value&key=value...]
:url: https://pypi.org/project/psycopg/
``psycopg`` is the package and module name for version 3 of the ``psycopg``
database driver, formerly known as ``psycopg2``. This driver is different
enough from its ``psycopg2`` predecessor that SQLAlchemy supports it
via a totally separate dialect; support for ``psycopg2`` is expected to remain
for as long as that package continues to function for modern Python versions,
and also remains the default dialect for the ``postgresql://`` dialect
series.
The SQLAlchemy ``psycopg`` dialect provides both a sync and an async
implementation under the same dialect name. The proper version is
selected depending on how the engine is created:
* calling :func:`_sa.create_engine` with ``postgresql+psycopg://...`` will
automatically select the sync version, e.g.::
from sqlalchemy import create_engine
sync_engine = create_engine(
"postgresql+psycopg://scott:tiger@localhost/test"
)
* calling :func:`_asyncio.create_async_engine` with
``postgresql+psycopg://...`` will automatically select the async version,
e.g.::
from sqlalchemy.ext.asyncio import create_async_engine
asyncio_engine = create_async_engine(
"postgresql+psycopg://scott:tiger@localhost/test"
)
The asyncio version of the dialect may also be specified explicitly using the
``psycopg_async`` suffix, as::
from sqlalchemy.ext.asyncio import create_async_engine
asyncio_engine = create_async_engine(
"postgresql+psycopg_async://scott:tiger@localhost/test"
)
.. seealso::
:ref:`postgresql_psycopg2` - The SQLAlchemy ``psycopg``
dialect shares most of its behavior with the ``psycopg2`` dialect.
Further documentation is available there.
Using psycopg Connection Pooling
--------------------------------
The ``psycopg`` driver provides its own connection pool implementation that
may be used in place of SQLAlchemy's pooling functionality.
This pool implementation provides support for fixed and dynamic pool sizes
(including automatic downsizing for unused connections), connection health
pre-checks, and support for both synchronous and asynchronous code
environments.
Here is an example that uses the sync version of the pool, using
``psycopg_pool >= 3.3`` that introduces support for ``close_returns=True``::
import psycopg_pool
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
# Create a psycopg_pool connection pool
my_pool = psycopg_pool.ConnectionPool(
conninfo="postgresql://scott:tiger@localhost/test",
close_returns=True, # Return "closed" active connections to the pool
# ... other pool parameters as desired ...
)
# Create an engine that uses the connection pool to get a connection
engine = create_engine(
url="postgresql+psycopg://", # Only need the dialect now
poolclass=NullPool, # Disable SQLAlchemy's default connection pool
creator=my_pool.getconn, # Use Psycopg 3 connection pool to obtain connections
)
Similarly an the async example::
import psycopg_pool
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool
async def define_engine():
# Create a psycopg_pool connection pool
my_pool = psycopg_pool.AsyncConnectionPool(
conninfo="postgresql://scott:tiger@localhost/test",
open=False, # See comment below
close_returns=True, # Return "closed" active connections to the pool
# ... other pool parameters as desired ...
)
# Must explicitly open AsyncConnectionPool outside constructor
# https://www.psycopg.org/psycopg3/docs/api/pool.html#psycopg_pool.AsyncConnectionPool
await my_pool.open()
# Create an engine that uses the connection pool to get a connection
engine = create_async_engine(
url="postgresql+psycopg://", # Only need the dialect now
poolclass=NullPool, # Disable SQLAlchemy's default connection pool
async_creator=my_pool.getconn, # Use Psycopg 3 connection pool to obtain connections
)
return engine, my_pool
The resulting engine may then be used normally. Internally, Psycopg 3 handles
connection pooling::
with engine.connect() as conn:
print(conn.scalar(text("select 42")))
.. seealso::
`Connection pools <https://www.psycopg.org/psycopg3/docs/advanced/pool.html>`_ -
the Psycopg 3 documentation for ``psycopg_pool.ConnectionPool``.
`Example for older version of psycopg_pool
<https://github.com/sqlalchemy/sqlalchemy/discussions/12522#discussioncomment-13024666>`_ -
An example about using the ``psycopg_pool<3.3`` that did not have the
``close_returns``` parameter.
Using a different Cursor class
------------------------------
One of the differences between ``psycopg`` and the older ``psycopg2``
is how bound parameters are handled: ``psycopg2`` would bind them
client side, while ``psycopg`` by default will bind them server side.
It's possible to configure ``psycopg`` to do client side binding by
specifying the ``cursor_factory`` to be ``ClientCursor`` when creating
the engine::
from psycopg import ClientCursor
client_side_engine = create_engine(
"postgresql+psycopg://...",
connect_args={"cursor_factory": ClientCursor},
)
Similarly when using an async engine the ``AsyncClientCursor`` can be
specified::
from psycopg import AsyncClientCursor
client_side_engine = create_async_engine(
"postgresql+psycopg://...",
connect_args={"cursor_factory": AsyncClientCursor},
)
.. seealso::
`Client-side-binding cursors <https://www.psycopg.org/psycopg3/docs/advanced/cursors.html#client-side-binding-cursors>`_
""" # noqa
from __future__ import annotations
from collections import deque
import logging
import re
from typing import cast
from typing import TYPE_CHECKING
from . import ranges
from ._psycopg_common import _PGDialect_common_psycopg
from ._psycopg_common import _PGExecutionContext_common_psycopg
from .base import INTERVAL
from .base import PGCompiler
from .base import PGIdentifierPreparer
from .base import REGCONFIG
from .json import JSON
from .json import JSONB
from .json import JSONPathType
from .types import CITEXT
from ... import pool
from ... import util
from ...engine import AdaptedConnection
from ...sql import sqltypes
from ...util.concurrency import await_fallback
from ...util.concurrency import await_only
if TYPE_CHECKING:
from typing import Iterable
from psycopg import AsyncConnection
logger = logging.getLogger("sqlalchemy.dialects.postgresql")
class _PGString(sqltypes.String):
render_bind_cast = True
class _PGREGCONFIG(REGCONFIG):
render_bind_cast = True
class _PGJSON(JSON):
def bind_processor(self, dialect):
return self._make_bind_processor(None, dialect._psycopg_Json)
def result_processor(self, dialect, coltype):
return None
class _PGJSONB(JSONB):
def bind_processor(self, dialect):
return self._make_bind_processor(None, dialect._psycopg_Jsonb)
def result_processor(self, dialect, coltype):
return None
class _PGJSONIntIndexType(sqltypes.JSON.JSONIntIndexType):
__visit_name__ = "json_int_index"
render_bind_cast = True
class _PGJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
__visit_name__ = "json_str_index"
render_bind_cast = True
class _PGJSONPathType(JSONPathType):
pass
class _PGInterval(INTERVAL):
render_bind_cast = True
class _PGTimeStamp(sqltypes.DateTime):
render_bind_cast = True
class _PGDate(sqltypes.Date):
render_bind_cast = True
class _PGTime(sqltypes.Time):
render_bind_cast = True
class _PGInteger(sqltypes.Integer):
render_bind_cast = True
class _PGSmallInteger(sqltypes.SmallInteger):
render_bind_cast = True
class _PGNullType(sqltypes.NullType):
render_bind_cast = True
class _PGBigInteger(sqltypes.BigInteger):
render_bind_cast = True
class _PGBoolean(sqltypes.Boolean):
render_bind_cast = True
class _PsycopgRange(ranges.AbstractSingleRangeImpl):
def bind_processor(self, dialect):
psycopg_Range = cast(PGDialect_psycopg, dialect)._psycopg_Range
def to_range(value):
if isinstance(value, ranges.Range):
value = psycopg_Range(
value.lower, value.upper, value.bounds, value.empty
)
return value
return to_range
def result_processor(self, dialect, coltype):
def to_range(value):
if value is not None:
value = ranges.Range(
value._lower,
value._upper,
bounds=value._bounds if value._bounds else "[)",
empty=not value._bounds,
)
return value
return to_range
class _PsycopgMultiRange(ranges.AbstractMultiRangeImpl):
def bind_processor(self, dialect):
psycopg_Range = cast(PGDialect_psycopg, dialect)._psycopg_Range
psycopg_Multirange = cast(
PGDialect_psycopg, dialect
)._psycopg_Multirange
NoneType = type(None)
def to_range(value):
if isinstance(value, (str, NoneType, psycopg_Multirange)):
return value
return psycopg_Multirange(
[
psycopg_Range(
element.lower,
element.upper,
element.bounds,
element.empty,
)
for element in cast("Iterable[ranges.Range]", value)
]
)
return to_range
def result_processor(self, dialect, coltype):
def to_range(value):
if value is None:
return None
else:
return ranges.MultiRange(
ranges.Range(
elem._lower,
elem._upper,
bounds=elem._bounds if elem._bounds else "[)",
empty=not elem._bounds,
)
for elem in value
)
return to_range
class PGExecutionContext_psycopg(_PGExecutionContext_common_psycopg):
pass
class PGCompiler_psycopg(PGCompiler):
pass
class PGIdentifierPreparer_psycopg(PGIdentifierPreparer):
pass
def _log_notices(diagnostic):
logger.info("%s: %s", diagnostic.severity, diagnostic.message_primary)
class PGDialect_psycopg(_PGDialect_common_psycopg):
driver = "psycopg"
supports_statement_cache = True
supports_server_side_cursors = True
default_paramstyle = "pyformat"
supports_sane_multi_rowcount = True
execution_ctx_cls = PGExecutionContext_psycopg
statement_compiler = PGCompiler_psycopg
preparer = PGIdentifierPreparer_psycopg
psycopg_version = (0, 0)
_has_native_hstore = True
_psycopg_adapters_map = None
colspecs = util.update_copy(
_PGDialect_common_psycopg.colspecs,
{
sqltypes.String: _PGString,
REGCONFIG: _PGREGCONFIG,
JSON: _PGJSON,
CITEXT: CITEXT,
sqltypes.JSON: _PGJSON,
JSONB: _PGJSONB,
sqltypes.JSON.JSONPathType: _PGJSONPathType,
sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
sqltypes.Interval: _PGInterval,
INTERVAL: _PGInterval,
sqltypes.Date: _PGDate,
sqltypes.DateTime: _PGTimeStamp,
sqltypes.Time: _PGTime,
sqltypes.Integer: _PGInteger,
sqltypes.SmallInteger: _PGSmallInteger,
sqltypes.BigInteger: _PGBigInteger,
ranges.AbstractSingleRange: _PsycopgRange,
ranges.AbstractMultiRange: _PsycopgMultiRange,
},
)
def __init__(self, **kwargs):
super().__init__(**kwargs)
if self.dbapi:
m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
if m:
self.psycopg_version = tuple(
int(x) for x in m.group(1, 2, 3) if x is not None
)
if self.psycopg_version < (3, 0, 2):
raise ImportError(
"psycopg version 3.0.2 or higher is required."
)
from psycopg.adapt import AdaptersMap
self._psycopg_adapters_map = adapters_map = AdaptersMap(
self.dbapi.adapters
)
if self._native_inet_types is False:
import psycopg.types.string
adapters_map.register_loader(
"inet", psycopg.types.string.TextLoader
)
adapters_map.register_loader(
"cidr", psycopg.types.string.TextLoader
)
if self._json_deserializer:
from psycopg.types.json import set_json_loads
set_json_loads(self._json_deserializer, adapters_map)
if self._json_serializer:
from psycopg.types.json import set_json_dumps
set_json_dumps(self._json_serializer, adapters_map)
def create_connect_args(self, url):
# see https://github.com/psycopg/psycopg/issues/83
cargs, cparams = super().create_connect_args(url)
if self._psycopg_adapters_map:
cparams["context"] = self._psycopg_adapters_map
if self.client_encoding is not None:
cparams["client_encoding"] = self.client_encoding
return cargs, cparams
def _type_info_fetch(self, connection, name):
from psycopg.types import TypeInfo
return TypeInfo.fetch(connection.connection.driver_connection, name)
def initialize(self, connection):
super().initialize(connection)
# PGDialect.initialize() checks server version for <= 8.2 and sets
# this flag to False if so
if not self.insert_returning:
self.insert_executemany_returning = False
# HSTORE can't be registered until we have a connection so that
# we can look up its OID, so we set up this adapter in
# initialize()
if self.use_native_hstore:
info = self._type_info_fetch(connection, "hstore")
self._has_native_hstore = info is not None
if self._has_native_hstore:
from psycopg.types.hstore import register_hstore
# register the adapter for connections made subsequent to
# this one
assert self._psycopg_adapters_map
register_hstore(info, self._psycopg_adapters_map)
# register the adapter for this connection
assert connection.connection
register_hstore(info, connection.connection.driver_connection)
@classmethod
def import_dbapi(cls):
import psycopg
return psycopg
@classmethod
def get_async_dialect_cls(cls, url):
return PGDialectAsync_psycopg
@util.memoized_property
def _isolation_lookup(self):
return {
"READ COMMITTED": self.dbapi.IsolationLevel.READ_COMMITTED,
"READ UNCOMMITTED": self.dbapi.IsolationLevel.READ_UNCOMMITTED,
"REPEATABLE READ": self.dbapi.IsolationLevel.REPEATABLE_READ,
"SERIALIZABLE": self.dbapi.IsolationLevel.SERIALIZABLE,
}
@util.memoized_property
def _psycopg_Json(self):
from psycopg.types import json
return json.Json
@util.memoized_property
def _psycopg_Jsonb(self):
from psycopg.types import json
return json.Jsonb
@util.memoized_property
def _psycopg_TransactionStatus(self):
from psycopg.pq import TransactionStatus
return TransactionStatus
@util.memoized_property
def _psycopg_Range(self):
from psycopg.types.range import Range
return Range
@util.memoized_property
def _psycopg_Multirange(self):
from psycopg.types.multirange import Multirange
return Multirange
def _do_isolation_level(self, connection, autocommit, isolation_level):
connection.autocommit = autocommit
connection.isolation_level = isolation_level
def get_isolation_level(self, dbapi_connection):
status_before = dbapi_connection.info.transaction_status
value = super().get_isolation_level(dbapi_connection)
# don't rely on psycopg providing enum symbols, compare with
# eq/ne
if status_before == self._psycopg_TransactionStatus.IDLE:
dbapi_connection.rollback()
return value
def set_isolation_level(self, dbapi_connection, level):
if level == "AUTOCOMMIT":
self._do_isolation_level(
dbapi_connection, autocommit=True, isolation_level=None
)
else:
self._do_isolation_level(
dbapi_connection,
autocommit=False,
isolation_level=self._isolation_lookup[level],
)
def set_readonly(self, connection, value):
connection.read_only = value
def get_readonly(self, connection):
return connection.read_only
def on_connect(self):
def notices(conn):
conn.add_notice_handler(_log_notices)
fns = [notices]
if self.isolation_level is not None:
def on_connect(conn):
self.set_isolation_level(conn, self.isolation_level)
fns.append(on_connect)
# fns always has the notices function
def on_connect(conn):
for fn in fns:
fn(conn)
return on_connect
def is_disconnect(self, e, connection, cursor):
if isinstance(e, self.dbapi.Error) and connection is not None:
if connection.closed or connection.broken:
return True
return False
def _twophase_idle_check(self, dbapi_conn):
# don't rely on psycopg providing enum symbols, compare with eq/ne
return (
dbapi_conn.info.transaction_status
== self._psycopg_TransactionStatus.IDLE
)
@util.memoized_property
def _dialect_specific_select_one(self):
return ";"
class AsyncAdapt_psycopg_cursor:
__slots__ = ("_cursor", "await_", "_rows")
_psycopg_ExecStatus = None
def __init__(self, cursor, await_) -> None:
self._cursor = cursor
self.await_ = await_
self._rows = deque()
def __getattr__(self, name):
return getattr(self._cursor, name)
@property
def arraysize(self):
return self._cursor.arraysize
@arraysize.setter
def arraysize(self, value):
self._cursor.arraysize = value
async def _async_soft_close(self) -> None:
return
def close(self):
self._rows.clear()
# Normal cursor just call _close() in a non-sync way.
self._cursor._close()
def execute(self, query, params=None, **kw):
result = self.await_(self._cursor.execute(query, params, **kw))
# sqlalchemy result is not async, so need to pull all rows here
res = self._cursor.pgresult
# don't rely on psycopg providing enum symbols, compare with
# eq/ne
if res and res.status == self._psycopg_ExecStatus.TUPLES_OK:
rows = self.await_(self._cursor.fetchall())
self._rows = deque(rows)
return result
def executemany(self, query, params_seq):
return self.await_(self._cursor.executemany(query, params_seq))
def __iter__(self):
while self._rows:
yield self._rows.popleft()
def fetchone(self):
if self._rows:
return self._rows.popleft()
else:
return None
def fetchmany(self, size=None):
if size is None:
size = self._cursor.arraysize
rr = self._rows
return [rr.popleft() for _ in range(min(size, len(rr)))]
def fetchall(self):
retval = list(self._rows)
self._rows.clear()
return retval
class AsyncAdapt_psycopg_ss_cursor(AsyncAdapt_psycopg_cursor):
def execute(self, query, params=None, **kw):
self.await_(self._cursor.execute(query, params, **kw))
return self
def close(self):
self.await_(self._cursor.close())
def fetchone(self):
return self.await_(self._cursor.fetchone())
def fetchmany(self, size=0):
return self.await_(self._cursor.fetchmany(size))
def fetchall(self):
return self.await_(self._cursor.fetchall())
def __iter__(self):
iterator = self._cursor.__aiter__()
while True:
try:
yield self.await_(iterator.__anext__())
except StopAsyncIteration:
break
class AsyncAdapt_psycopg_connection(AdaptedConnection):
_connection: AsyncConnection
__slots__ = ()
await_ = staticmethod(await_only)
def __init__(self, connection) -> None:
self._connection = connection
def __getattr__(self, name):
return getattr(self._connection, name)
def execute(self, query, params=None, **kw):
cursor = self.await_(self._connection.execute(query, params, **kw))
return AsyncAdapt_psycopg_cursor(cursor, self.await_)
def cursor(self, *args, **kw):
cursor = self._connection.cursor(*args, **kw)
if hasattr(cursor, "name"):
return AsyncAdapt_psycopg_ss_cursor(cursor, self.await_)
else:
return AsyncAdapt_psycopg_cursor(cursor, self.await_)
def commit(self):
self.await_(self._connection.commit())
def rollback(self):
self.await_(self._connection.rollback())
def close(self):
self.await_(self._connection.close())
@property
def autocommit(self):
return self._connection.autocommit
@autocommit.setter
def autocommit(self, value):
self.set_autocommit(value)
def set_autocommit(self, value):
self.await_(self._connection.set_autocommit(value))
def set_isolation_level(self, value):
self.await_(self._connection.set_isolation_level(value))
def set_read_only(self, value):
self.await_(self._connection.set_read_only(value))
def set_deferrable(self, value):
self.await_(self._connection.set_deferrable(value))
def tpc_begin(self, xid):
return self.await_(self._connection.tpc_begin(xid))
def tpc_prepare(self):
return self.await_(self._connection.tpc_prepare())
def tpc_commit(self, xid=None):
return self.await_(self._connection.tpc_commit(xid))
def tpc_rollback(self, xid=None):
return self.await_(self._connection.tpc_rollback(xid))
def tpc_recover(self):
return self.await_(self._connection.tpc_recover())
class AsyncAdaptFallback_psycopg_connection(AsyncAdapt_psycopg_connection):
__slots__ = ()
await_ = staticmethod(await_fallback)
class PsycopgAdaptDBAPI:
def __init__(self, psycopg) -> None:
self.psycopg = psycopg
for k, v in self.psycopg.__dict__.items():
if k != "connect":
self.__dict__[k] = v
def connect(self, *arg, **kw):
async_fallback = kw.pop("async_fallback", False)
creator_fn = kw.pop(
"async_creator_fn", self.psycopg.AsyncConnection.connect
)
if util.asbool(async_fallback):
return AsyncAdaptFallback_psycopg_connection(
await_fallback(creator_fn(*arg, **kw))
)
else:
return AsyncAdapt_psycopg_connection(
await_only(creator_fn(*arg, **kw))
)
class PGDialectAsync_psycopg(PGDialect_psycopg):
is_async = True
supports_statement_cache = True
@classmethod
def import_dbapi(cls):
import psycopg
from psycopg.pq import ExecStatus
AsyncAdapt_psycopg_cursor._psycopg_ExecStatus = ExecStatus
return PsycopgAdaptDBAPI(psycopg)
@classmethod
def get_pool_class(cls, url):
async_fallback = url.query.get("async_fallback", False)
if util.asbool(async_fallback):
return pool.FallbackAsyncAdaptedQueuePool
else:
return pool.AsyncAdaptedQueuePool
def _type_info_fetch(self, connection, name):
from psycopg.types import TypeInfo
adapted = connection.connection
return adapted.await_(TypeInfo.fetch(adapted.driver_connection, name))
def _do_isolation_level(self, connection, autocommit, isolation_level):
connection.set_autocommit(autocommit)
connection.set_isolation_level(isolation_level)
def _do_autocommit(self, connection, value):
connection.set_autocommit(value)
def set_readonly(self, connection, value):
connection.set_read_only(value)
def set_deferrable(self, connection, value):
connection.set_deferrable(value)
def get_driver_connection(self, connection):
return connection._connection
dialect = PGDialect_psycopg
dialect_async = PGDialectAsync_psycopg