codekingpro/portable-devtools
114k
1# dialects/postgresql/pg8000.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors <see AUTHORS
3# file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9r"""
10.. dialect:: postgresql+pg8000
11 :name: pg8000
12 :dbapi: pg8000
13 :connectstring: postgresql+pg8000://user:password@host:port/dbname[?key=value&key=value...]
14 :url: https://pypi.org/project/pg8000/
15
16.. versionchanged:: 1.4 The pg8000 dialect has been updated for version
17 1.16.6 and higher, and is again part of SQLAlchemy's continuous integration
18 with full feature support.
19
20.. _pg8000_unicode:
21
22Unicode
23-------
24
25pg8000 will encode / decode string values between it and the server using the
26PostgreSQL ``client_encoding`` parameter; by default this is the value in
27the ``postgresql.conf`` file, which often defaults to ``SQL_ASCII``.
28Typically, this can be changed to ``utf-8``, as a more useful default::
29
30 #client_encoding = sql_ascii # actually, defaults to database
31 # encoding
32 client_encoding = utf8
33
34The ``client_encoding`` can be overridden for a session by executing the SQL:
35
36SET CLIENT_ENCODING TO 'utf8';
37
38SQLAlchemy will execute this SQL on all new connections based on the value
39passed to :func:`_sa.create_engine` using the ``client_encoding`` parameter::
40
41 engine = create_engine(
42 "postgresql+pg8000://user:pass@host/dbname", client_encoding='utf8')
43
44.. _pg8000_ssl:
45
46SSL Connections
47---------------
48
49pg8000 accepts a Python ``SSLContext`` object which may be specified using the
50:paramref:`_sa.create_engine.connect_args` dictionary::
51
52 import ssl
53 ssl_context = ssl.create_default_context()
54 engine = sa.create_engine(
55 "postgresql+pg8000://scott:tiger@192.168.0.199/test",
56 connect_args={"ssl_context": ssl_context},
57 )
58
59If the server uses an automatically-generated certificate that is self-signed
60or does not match the host name (as seen from the client), it may also be
61necessary to disable hostname checking::
62
63 import ssl
64 ssl_context = ssl.create_default_context()
65 ssl_context.check_hostname = False
66 ssl_context.verify_mode = ssl.CERT_NONE
67 engine = sa.create_engine(
68 "postgresql+pg8000://scott:tiger@192.168.0.199/test",
69 connect_args={"ssl_context": ssl_context},
70 )
71
72.. _pg8000_isolation_level:
73
74pg8000 Transaction Isolation Level
75-------------------------------------
76
77The pg8000 dialect offers the same isolation level settings as that
78of the :ref:`psycopg2 <psycopg2_isolation_level>` dialect:
79
80* ``READ COMMITTED``
81* ``READ UNCOMMITTED``
82* ``REPEATABLE READ``
83* ``SERIALIZABLE``
84* ``AUTOCOMMIT``
85
86.. seealso::
87
88 :ref:`postgresql_isolation_level`
89
90 :ref:`psycopg2_isolation_level`
91
92
93""" # noqa
94import decimal
95import re
96
97from . import ranges
98from .array import ARRAY as PGARRAY
99from .base import _DECIMAL_TYPES
100from .base import _FLOAT_TYPES
101from .base import _INT_TYPES
102from .base import ENUM
103from .base import INTERVAL
104from .base import PGCompiler
105from .base import PGDialect
106from .base import PGExecutionContext
107from .base import PGIdentifierPreparer
108from .json import JSON
109from .json import JSONB
110from .json import JSONPathType
111from .pg_catalog import _SpaceVector
112from .pg_catalog import OIDVECTOR
113from .types import CITEXT
114from ... import exc
115from ... import util
116from ...engine import processors
117from ...sql import sqltypes
118from ...sql.elements import quoted_name
119
120
121class _PGString(sqltypes.String):
122 render_bind_cast = True
123
124
125class _PGNumeric(sqltypes.Numeric):
126 render_bind_cast = True
127
128 def result_processor(self, dialect, coltype):
129 if self.asdecimal:
130 if coltype in _FLOAT_TYPES:
131 return processors.to_decimal_processor_factory(
132 decimal.Decimal, self._effective_decimal_return_scale
133 )
134 elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
135 # pg8000 returns Decimal natively for 1700
136 return None
137 else:
138 raise exc.InvalidRequestError(
139 "Unknown PG numeric type: %d" % coltype
140 )
141 else:
142 if coltype in _FLOAT_TYPES:
143 # pg8000 returns float natively for 701
144 return None
145 elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
146 return processors.to_float
147 else:
148 raise exc.InvalidRequestError(
149 "Unknown PG numeric type: %d" % coltype
150 )
151
152
153class _PGFloat(_PGNumeric, sqltypes.Float):
154 __visit_name__ = "float"
155 render_bind_cast = True
156
157
158class _PGNumericNoBind(_PGNumeric):
159 def bind_processor(self, dialect):
160 return None
161
162
163class _PGJSON(JSON):
164 render_bind_cast = True
165
166 def result_processor(self, dialect, coltype):
167 return None
168
169
170class _PGJSONB(JSONB):
171 render_bind_cast = True
172
173 def result_processor(self, dialect, coltype):
174 return None
175
176
177class _PGJSONIndexType(sqltypes.JSON.JSONIndexType):
178 def get_dbapi_type(self, dbapi):
179 raise NotImplementedError("should not be here")
180
181
182class _PGJSONIntIndexType(sqltypes.JSON.JSONIntIndexType):
183 __visit_name__ = "json_int_index"
184
185 render_bind_cast = True
186
187
188class _PGJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
189 __visit_name__ = "json_str_index"
190
191 render_bind_cast = True
192
193
194class _PGJSONPathType(JSONPathType):
195 pass
196
197 # DBAPI type 1009
198
199
200class _PGEnum(ENUM):
201 def get_dbapi_type(self, dbapi):
202 return dbapi.UNKNOWN
203
204
205class _PGInterval(INTERVAL):
206 render_bind_cast = True
207
208 def get_dbapi_type(self, dbapi):
209 return dbapi.INTERVAL
210
211 @classmethod
212 def adapt_emulated_to_native(cls, interval, **kw):
213 return _PGInterval(precision=interval.second_precision)
214
215
216class _PGTimeStamp(sqltypes.DateTime):
217 render_bind_cast = True
218
219
220class _PGDate(sqltypes.Date):
221 render_bind_cast = True
222
223
224class _PGTime(sqltypes.Time):
225 render_bind_cast = True
226
227
228class _PGInteger(sqltypes.Integer):
229 render_bind_cast = True
230
231
232class _PGSmallInteger(sqltypes.SmallInteger):
233 render_bind_cast = True
234
235
236class _PGNullType(sqltypes.NullType):
237 pass
238
239
240class _PGBigInteger(sqltypes.BigInteger):
241 render_bind_cast = True
242
243
244class _PGBoolean(sqltypes.Boolean):
245 render_bind_cast = True
246
247
248class _PGARRAY(PGARRAY):
249 render_bind_cast = True
250
251
252class _PGOIDVECTOR(_SpaceVector, OIDVECTOR):
253 pass
254
255
256class _Pg8000Range(ranges.AbstractSingleRangeImpl):
257 def bind_processor(self, dialect):
258 pg8000_Range = dialect.dbapi.Range
259
260 def to_range(value):
261 if isinstance(value, ranges.Range):
262 value = pg8000_Range(
263 value.lower, value.upper, value.bounds, value.empty
264 )
265 return value
266
267 return to_range
268
269 def result_processor(self, dialect, coltype):
270 def to_range(value):
271 if value is not None:
272 value = ranges.Range(
273 value.lower,
274 value.upper,
275 bounds=value.bounds,
276 empty=value.is_empty,
277 )
278 return value
279
280 return to_range
281
282
283class _Pg8000MultiRange(ranges.AbstractMultiRangeImpl):
284 def bind_processor(self, dialect):
285 pg8000_Range = dialect.dbapi.Range
286
287 def to_multirange(value):
288 if isinstance(value, list):
289 mr = []
290 for v in value:
291 if isinstance(v, ranges.Range):
292 mr.append(
293 pg8000_Range(v.lower, v.upper, v.bounds, v.empty)
294 )
295 else:
296 mr.append(v)
297 return mr
298 else:
299 return value
300
301 return to_multirange
302
303 def result_processor(self, dialect, coltype):
304 def to_multirange(value):
305 if value is None:
306 return None
307 else:
308 return ranges.MultiRange(
309 ranges.Range(
310 v.lower, v.upper, bounds=v.bounds, empty=v.is_empty
311 )
312 for v in value
313 )
314
315 return to_multirange
316
317
318_server_side_id = util.counter()
319
320
321class PGExecutionContext_pg8000(PGExecutionContext):
322 def create_server_side_cursor(self):
323 ident = "c_%s_%s" % (hex(id(self))[2:], hex(_server_side_id())[2:])
324 return ServerSideCursor(self._dbapi_connection.cursor(), ident)
325
326 def pre_exec(self):
327 if not self.compiled:
328 return
329
330
331class ServerSideCursor:
332 server_side = True
333
334 def __init__(self, cursor, ident):
335 self.ident = ident
336 self.cursor = cursor
337
338 @property
339 def connection(self):
340 return self.cursor.connection
341
342 @property
343 def rowcount(self):
344 return self.cursor.rowcount
345
346 @property
347 def description(self):
348 return self.cursor.description
349
350 def execute(self, operation, args=(), stream=None):
351 op = "DECLARE " + self.ident + " NO SCROLL CURSOR FOR " + operation
352 self.cursor.execute(op, args, stream=stream)
353 return self
354
355 def executemany(self, operation, param_sets):
356 self.cursor.executemany(operation, param_sets)
357 return self
358
359 def fetchone(self):
360 self.cursor.execute("FETCH FORWARD 1 FROM " + self.ident)
361 return self.cursor.fetchone()
362
363 def fetchmany(self, num=None):
364 if num is None:
365 return self.fetchall()
366 else:
367 self.cursor.execute(
368 "FETCH FORWARD " + str(int(num)) + " FROM " + self.ident
369 )
370 return self.cursor.fetchall()
371
372 def fetchall(self):
373 self.cursor.execute("FETCH FORWARD ALL FROM " + self.ident)
374 return self.cursor.fetchall()
375
376 def close(self):
377 self.cursor.execute("CLOSE " + self.ident)
378 self.cursor.close()
379
380 def setinputsizes(self, *sizes):
381 self.cursor.setinputsizes(*sizes)
382
383 def setoutputsize(self, size, column=None):
384 pass
385
386
387class PGCompiler_pg8000(PGCompiler):
388 def visit_mod_binary(self, binary, operator, **kw):
389 return (
390 self.process(binary.left, **kw)
391 + " %% "
392 + self.process(binary.right, **kw)
393 )
394
395
396class PGIdentifierPreparer_pg8000(PGIdentifierPreparer):
397 def __init__(self, *args, **kwargs):
398 PGIdentifierPreparer.__init__(self, *args, **kwargs)
399 self._double_percents = False
400
401
402class PGDialect_pg8000(PGDialect):
403 driver = "pg8000"
404 supports_statement_cache = True
405
406 supports_unicode_statements = True
407
408 supports_unicode_binds = True
409
410 default_paramstyle = "format"
411 supports_sane_multi_rowcount = True
412 execution_ctx_cls = PGExecutionContext_pg8000
413 statement_compiler = PGCompiler_pg8000
414 preparer = PGIdentifierPreparer_pg8000
415 supports_server_side_cursors = True
416
417 render_bind_cast = True
418
419 # reversed as of pg8000 1.16.6. 1.16.5 and lower
420 # are no longer compatible
421 description_encoding = None
422 # description_encoding = "use_encoding"
423
424 colspecs = util.update_copy(
425 PGDialect.colspecs,
426 {
427 sqltypes.String: _PGString,
428 sqltypes.Numeric: _PGNumericNoBind,
429 sqltypes.Float: _PGFloat,
430 sqltypes.JSON: _PGJSON,
431 sqltypes.Boolean: _PGBoolean,
432 sqltypes.NullType: _PGNullType,
433 JSONB: _PGJSONB,
434 CITEXT: CITEXT,
435 sqltypes.JSON.JSONPathType: _PGJSONPathType,
436 sqltypes.JSON.JSONIndexType: _PGJSONIndexType,
437 sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
438 sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
439 sqltypes.Interval: _PGInterval,
440 INTERVAL: _PGInterval,
441 sqltypes.DateTime: _PGTimeStamp,
442 sqltypes.DateTime: _PGTimeStamp,
443 sqltypes.Date: _PGDate,
444 sqltypes.Time: _PGTime,
445 sqltypes.Integer: _PGInteger,
446 sqltypes.SmallInteger: _PGSmallInteger,
447 sqltypes.BigInteger: _PGBigInteger,
448 sqltypes.Enum: _PGEnum,
449 sqltypes.ARRAY: _PGARRAY,
450 OIDVECTOR: _PGOIDVECTOR,
451 ranges.INT4RANGE: _Pg8000Range,
452 ranges.INT8RANGE: _Pg8000Range,
453 ranges.NUMRANGE: _Pg8000Range,
454 ranges.DATERANGE: _Pg8000Range,
455 ranges.TSRANGE: _Pg8000Range,
456 ranges.TSTZRANGE: _Pg8000Range,
457 ranges.INT4MULTIRANGE: _Pg8000MultiRange,
458 ranges.INT8MULTIRANGE: _Pg8000MultiRange,
459 ranges.NUMMULTIRANGE: _Pg8000MultiRange,
460 ranges.DATEMULTIRANGE: _Pg8000MultiRange,
461 ranges.TSMULTIRANGE: _Pg8000MultiRange,
462 ranges.TSTZMULTIRANGE: _Pg8000MultiRange,
463 },
464 )
465
466 def __init__(self, client_encoding=None, **kwargs):
467 PGDialect.__init__(self, **kwargs)
468 self.client_encoding = client_encoding
469
470 if self._dbapi_version < (1, 16, 6):
471 raise NotImplementedError("pg8000 1.16.6 or greater is required")
472
473 if self._native_inet_types:
474 raise NotImplementedError(
475 "The pg8000 dialect does not fully implement "
476 "ipaddress type handling; INET is supported by default, "
477 "CIDR is not"
478 )
479
480 @util.memoized_property
481 def _dbapi_version(self):
482 if self.dbapi and hasattr(self.dbapi, "__version__"):
483 return tuple(
484 [
485 int(x)
486 for x in re.findall(
487 r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
488 )
489 ]
490 )
491 else:
492 return (99, 99, 99)
493
494 @classmethod
495 def import_dbapi(cls):
496 return __import__("pg8000")
497
498 def create_connect_args(self, url):
499 opts = url.translate_connect_args(username="user")
500 if "port" in opts:
501 opts["port"] = int(opts["port"])
502 opts.update(url.query)
503 return ([], opts)
504
505 def is_disconnect(self, e, connection, cursor):
506 if isinstance(e, self.dbapi.InterfaceError) and "network error" in str(
507 e
508 ):
509 # new as of pg8000 1.19.0 for broken connections
510 return True
511
512 # connection was closed normally
513 return "connection is closed" in str(e)
514
515 def get_isolation_level_values(self, dbapi_connection):
516 return (
517 "AUTOCOMMIT",
518 "READ COMMITTED",
519 "READ UNCOMMITTED",
520 "REPEATABLE READ",
521 "SERIALIZABLE",
522 )
523
524 def set_isolation_level(self, dbapi_connection, level):
525 level = level.replace("_", " ")
526
527 if level == "AUTOCOMMIT":
528 dbapi_connection.autocommit = True
529 else:
530 dbapi_connection.autocommit = False
531 cursor = dbapi_connection.cursor()
532 cursor.execute(
533 "SET SESSION CHARACTERISTICS AS TRANSACTION "
534 f"ISOLATION LEVEL {level}"
535 )
536 cursor.execute("COMMIT")
537 cursor.close()
538
539 def set_readonly(self, connection, value):
540 cursor = connection.cursor()
541 try:
542 cursor.execute(
543 "SET SESSION CHARACTERISTICS AS TRANSACTION %s"
544 % ("READ ONLY" if value else "READ WRITE")
545 )
546 cursor.execute("COMMIT")
547 finally:
548 cursor.close()
549
550 def get_readonly(self, connection):
551 cursor = connection.cursor()
552 try:
553 cursor.execute("show transaction_read_only")
554 val = cursor.fetchone()[0]
555 finally:
556 cursor.close()
557
558 return val == "on"
559
560 def set_deferrable(self, connection, value):
561 cursor = connection.cursor()
562 try:
563 cursor.execute(
564 "SET SESSION CHARACTERISTICS AS TRANSACTION %s"
565 % ("DEFERRABLE" if value else "NOT DEFERRABLE")
566 )
567 cursor.execute("COMMIT")
568 finally:
569 cursor.close()
570
571 def get_deferrable(self, connection):
572 cursor = connection.cursor()
573 try:
574 cursor.execute("show transaction_deferrable")
575 val = cursor.fetchone()[0]
576 finally:
577 cursor.close()
578
579 return val == "on"
580
581 def _set_client_encoding(self, dbapi_connection, client_encoding):
582 cursor = dbapi_connection.cursor()
583 cursor.execute(
584 f"""SET CLIENT_ENCODING TO '{
585 client_encoding.replace("'", "''")
586 }'"""
587 )
588 cursor.execute("COMMIT")
589 cursor.close()
590
591 def do_begin_twophase(self, connection, xid):
592 connection.connection.tpc_begin((0, xid, ""))
593
594 def do_prepare_twophase(self, connection, xid):
595 connection.connection.tpc_prepare()
596
597 def do_rollback_twophase(
598 self, connection, xid, is_prepared=True, recover=False
599 ):
600 connection.connection.tpc_rollback((0, xid, ""))
601
602 def do_commit_twophase(
603 self, connection, xid, is_prepared=True, recover=False
604 ):
605 connection.connection.tpc_commit((0, xid, ""))
606
607 def do_recover_twophase(self, connection):
608 return [row[1] for row in connection.connection.tpc_recover()]
609
610 def on_connect(self):
611 fns = []
612
613 def on_connect(conn):
614 conn.py_types[quoted_name] = conn.py_types[str]
615
616 fns.append(on_connect)
617
618 if self.client_encoding is not None:
619
620 def on_connect(conn):
621 self._set_client_encoding(conn, self.client_encoding)
622
623 fns.append(on_connect)
624
625 if self._native_inet_types is False:
626
627 def on_connect(conn):
628 # inet
629 conn.register_in_adapter(869, lambda s: s)
630
631 # cidr
632 conn.register_in_adapter(650, lambda s: s)
633
634 fns.append(on_connect)
635
636 if self._json_deserializer:
637
638 def on_connect(conn):
639 # json
640 conn.register_in_adapter(114, self._json_deserializer)
641
642 # jsonb
643 conn.register_in_adapter(3802, self._json_deserializer)
644
645 fns.append(on_connect)
646
647 if len(fns) > 0:
648
649 def on_connect(conn):
650 for fn in fns:
651 fn(conn)
652
653 return on_connect
654 else:
655 return None
656
657 @util.memoized_property
658 def _dialect_specific_select_one(self):
659 return ";"
660
661
662dialect = PGDialect_pg8000
663