1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
# Copyright (C) 2005-2023 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 __future__ import annotations
import decimal
from .array import ARRAY as PGARRAY
from .base import _DECIMAL_TYPES
from .base import _FLOAT_TYPES
from .base import _INT_TYPES
from .base import PGDialect
from .base import PGExecutionContext
from .hstore import HSTORE
from .pg_catalog import _SpaceVector
from .pg_catalog import INT2VECTOR
from .pg_catalog import OIDVECTOR
from ... import exc
from ... import types as sqltypes
from ... import util
from ...engine import processors
_server_side_id = util.counter()
class _PsycopgNumeric(sqltypes.Numeric):
def bind_processor(self, dialect):
return None
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:
# psycopg returns Decimal natively for 1700
return None
else:
raise exc.InvalidRequestError(
"Unknown PG numeric type: %d" % coltype
)
else:
if coltype in _FLOAT_TYPES:
# psycopg 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 _PsycopgFloat(_PsycopgNumeric):
__visit_name__ = "float"
class _PsycopgHStore(HSTORE):
def bind_processor(self, dialect):
if dialect._has_native_hstore:
return None
else:
return super().bind_processor(dialect)
def result_processor(self, dialect, coltype):
if dialect._has_native_hstore:
return None
else:
return super().result_processor(dialect, coltype)
class _PsycopgARRAY(PGARRAY):
render_bind_cast = True
class _PsycopgINT2VECTOR(_SpaceVector, INT2VECTOR):
pass
class _PsycopgOIDVECTOR(_SpaceVector, OIDVECTOR):
pass
class _PGExecutionContext_common_psycopg(PGExecutionContext):
def create_server_side_cursor(self):
# use server-side cursors:
# psycopg
# https://www.psycopg.org/psycopg3/docs/advanced/cursors.html#server-side-cursors
# psycopg2
# https://www.psycopg.org/docs/usage.html#server-side-cursors
ident = "c_%s_%s" % (hex(id(self))[2:], hex(_server_side_id())[2:])
return self._dbapi_connection.cursor(ident)
class _PGDialect_common_psycopg(PGDialect):
supports_statement_cache = True
supports_server_side_cursors = True
default_paramstyle = "pyformat"
_has_native_hstore = True
colspecs = util.update_copy(
PGDialect.colspecs,
{
sqltypes.Numeric: _PsycopgNumeric,
sqltypes.Float: _PsycopgFloat,
HSTORE: _PsycopgHStore,
sqltypes.ARRAY: _PsycopgARRAY,
INT2VECTOR: _PsycopgINT2VECTOR,
OIDVECTOR: _PsycopgOIDVECTOR,
},
)
def __init__(
self,
client_encoding=None,
use_native_hstore=True,
**kwargs,
):
PGDialect.__init__(self, **kwargs)
if not use_native_hstore:
self._has_native_hstore = False
self.use_native_hstore = use_native_hstore
self.client_encoding = client_encoding
def create_connect_args(self, url):
opts = url.translate_connect_args(username="user", database="dbname")
is_multihost = False
if "host" in url.query:
is_multihost = isinstance(url.query["host"], (list, tuple))
if opts or url.query:
if not opts:
opts = {}
if "port" in opts:
opts["port"] = int(opts["port"])
opts.update(url.query)
if is_multihost:
hosts, ports = zip(
*[
token.split(":") if ":" in token else (token, "")
for token in url.query["host"]
]
)
opts["host"] = ",".join(hosts)
if "port" in opts:
raise exc.ArgumentError(
"Can't mix 'multihost' formats together; use "
'"host=h1,h2,h3&port=p1,p2,p3" or '
'"host=h1:p1&host=h2:p2&host=h3:p3" separately'
)
opts["port"] = ",".join(ports)
return ([], opts)
else:
# no connection arguments whatsoever; psycopg2.connect()
# requires that "dsn" be present as a blank string.
return ([""], opts)
def get_isolation_level_values(self, dbapi_connection):
return (
"AUTOCOMMIT",
"READ COMMITTED",
"READ UNCOMMITTED",
"REPEATABLE READ",
"SERIALIZABLE",
)
def set_deferrable(self, connection, value):
connection.deferrable = value
def get_deferrable(self, connection):
return connection.deferrable
def _do_autocommit(self, connection, value):
connection.autocommit = value
def do_ping(self, dbapi_connection):
cursor = None
before_autocommit = dbapi_connection.autocommit
if not before_autocommit:
dbapi_connection.autocommit = True
cursor = dbapi_connection.cursor()
try:
cursor.execute(self._dialect_specific_select_one)
finally:
cursor.close()
if not before_autocommit and not dbapi_connection.closed:
dbapi_connection.autocommit = before_autocommit
return True
|
