Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/postgresql/pg8000.py: 49%

1# dialects/postgresql/pg8000.py 

2# Copyright (C) 2005-2026 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 encoding 

31 client_encoding = utf8 

32 

33The ``client_encoding`` can be overridden for a session by executing the SQL: 

34 

35.. sourcecode:: sql 

36 

37 SET CLIENT_ENCODING TO 'utf8'; 

38 

39SQLAlchemy will execute this SQL on all new connections based on the value 

40passed to :func:`_sa.create_engine` using the ``client_encoding`` parameter:: 

41 

42 engine = create_engine( 

43 "postgresql+pg8000://user:pass@host/dbname", client_encoding="utf8" 

44 ) 

45 

46.. _pg8000_ssl: 

47 

48SSL Connections 

49--------------- 

50 

51pg8000 accepts a Python ``SSLContext`` object which may be specified using the 

52:paramref:`_sa.create_engine.connect_args` dictionary:: 

53 

54 import ssl 

55 

56 ssl_context = ssl.create_default_context() 

57 engine = sa.create_engine( 

58 "postgresql+pg8000://scott:tiger@192.168.0.199/test", 

59 connect_args={"ssl_context": ssl_context}, 

60 ) 

61 

62If the server uses an automatically-generated certificate that is self-signed 

63or does not match the host name (as seen from the client), it may also be 

64necessary to disable hostname checking:: 

65 

66 import ssl 

67 

68 ssl_context = ssl.create_default_context() 

69 ssl_context.check_hostname = False 

70 ssl_context.verify_mode = ssl.CERT_NONE 

71 engine = sa.create_engine( 

72 "postgresql+pg8000://scott:tiger@192.168.0.199/test", 

73 connect_args={"ssl_context": ssl_context}, 

74 ) 

75 

76.. _pg8000_isolation_level: 

77 

78pg8000 Transaction Isolation Level 

79------------------------------------- 

80 

81The pg8000 dialect offers the same isolation level settings as that 

82of the :ref:`psycopg2 <psycopg2_isolation_level>` dialect: 

83 

84* ``READ COMMITTED`` 

85* ``READ UNCOMMITTED`` 

86* ``REPEATABLE READ`` 

87* ``SERIALIZABLE`` 

88* ``AUTOCOMMIT`` 

89 

90.. seealso:: 

91 

92 :ref:`postgresql_isolation_level` 

93 

94 :ref:`psycopg2_isolation_level` 

95 

96 

97""" # noqa 

98 

99import decimal 

100 

101from . import ranges 

102from .array import ARRAY as PGARRAY 

103from .base import _DECIMAL_TYPES 

104from .base import _FLOAT_TYPES 

105from .base import _INT_TYPES 

106from .base import ENUM 

107from .base import INTERVAL 

108from .base import PGCompiler 

109from .base import PGDialect 

110from .base import PGExecutionContext 

111from .base import PGIdentifierPreparer 

112from .json import JSON 

113from .json import JSONB 

114from .json import JSONPathType 

115from .pg_catalog import _SpaceVector 

116from .pg_catalog import OIDVECTOR 

117from .types import CITEXT 

118from ... import exc 

119from ... import util 

120from ...engine import processors 

121from ...sql import sqltypes 

122from ...sql.elements import quoted_name 

123 

124 

125class _PGString(sqltypes.String): 

126 render_bind_cast = True 

127 

128 

129class _PGNumericCommon(sqltypes.NumericCommon): 

130 render_bind_cast = True 

131 

132 def result_processor(self, dialect, coltype): 

133 if self.asdecimal: 

134 if coltype in _FLOAT_TYPES: 

135 return processors.to_decimal_processor_factory( 

136 decimal.Decimal, self._effective_decimal_return_scale 

137 ) 

138 elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES: 

139 # pg8000 returns Decimal natively for 1700 

140 return None 

141 else: 

142 raise exc.InvalidRequestError( 

143 "Unknown PG numeric type: %d" % coltype 

144 ) 

145 else: 

146 if coltype in _FLOAT_TYPES: 

147 # pg8000 returns float natively for 701 

148 return None 

149 elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES: 

150 return processors.to_float 

151 else: 

152 raise exc.InvalidRequestError( 

153 "Unknown PG numeric type: %d" % coltype 

154 ) 

155 

156 

157class _PGNumeric(_PGNumericCommon, sqltypes.Numeric): 

158 pass 

159 

160 

161class _PGFloat(_PGNumericCommon, sqltypes.Float): 

162 pass 

163 

164 

165class _PGNumericNoBind(_PGNumeric): 

166 def bind_processor(self, dialect): 

167 return None 

168 

169 

170class _PGJSON(JSON): 

171 render_bind_cast = True 

172 

173 

174class _PGJSONB(JSONB): 

175 render_bind_cast = True 

176 

177 

178class _PGJSONIndexType(sqltypes.JSON.JSONIndexType): 

179 def get_dbapi_type(self, dbapi): 

180 raise NotImplementedError("should not be here") 

181 

182 

183class _PGJSONIntIndexType(sqltypes.JSON.JSONIntIndexType): 

184 __visit_name__ = "json_int_index" 

185 

186 render_bind_cast = True 

187 

188 

189class _PGJSONStrIndexType(sqltypes.JSON.JSONStrIndexType): 

190 __visit_name__ = "json_str_index" 

191 

192 render_bind_cast = True 

193 

194 

195class _PGJSONPathType(JSONPathType): 

196 pass 

197 

198 # DBAPI type 1009 

199 

200 

201class _PGEnum(ENUM): 

202 def get_dbapi_type(self, dbapi): 

203 return dbapi.UNKNOWN 

204 

205 

206class _PGInterval(INTERVAL): 

207 render_bind_cast = True 

208 

209 def get_dbapi_type(self, dbapi): 

210 return dbapi.INTERVAL 

211 

212 @classmethod 

213 def adapt_emulated_to_native(cls, interval, **kw): 

214 return _PGInterval(precision=interval.second_precision) 

215 

216 

217class _PGTimeStamp(sqltypes.DateTime): 

218 render_bind_cast = True 

219 

220 

221class _PGDate(sqltypes.Date): 

222 render_bind_cast = True 

223 

224 

225class _PGTime(sqltypes.Time): 

226 render_bind_cast = True 

227 

228 

229class _PGInteger(sqltypes.Integer): 

230 render_bind_cast = True 

231 

232 

233class _PGSmallInteger(sqltypes.SmallInteger): 

234 render_bind_cast = True 

235 

236 

237class _PGNullType(sqltypes.NullType): 

238 pass 

239 

240 

241class _PGBigInteger(sqltypes.BigInteger): 

242 render_bind_cast = True 

243 

244 

245class _PGBoolean(sqltypes.Boolean): 

246 render_bind_cast = True 

247 

248 

249class _PGARRAY(PGARRAY): 

250 render_bind_cast = True 

251 

252 

253class _PGOIDVECTOR(_SpaceVector, OIDVECTOR): 

254 pass 

255 

256 

257class _Pg8000Range(ranges.AbstractSingleRangeImpl): 

258 def bind_processor(self, dialect): 

259 pg8000_Range = dialect.dbapi.Range 

260 

261 def to_range(value): 

262 if isinstance(value, ranges.Range): 

263 value = pg8000_Range( 

264 value.lower, value.upper, value.bounds, value.empty 

265 ) 

266 return value 

267 

268 return to_range 

269 

270 def result_processor(self, dialect, coltype): 

271 def to_range(value): 

272 if value is not None: 

273 value = ranges.Range( 

274 value.lower, 

275 value.upper, 

276 bounds=value.bounds, 

277 empty=value.is_empty, 

278 ) 

279 return value 

280 

281 return to_range 

282 

283 

284class _Pg8000MultiRange(ranges.AbstractMultiRangeImpl): 

285 def bind_processor(self, dialect): 

286 pg8000_Range = dialect.dbapi.Range 

287 

288 def to_multirange(value): 

289 if isinstance(value, list): 

290 mr = [] 

291 for v in value: 

292 if isinstance(v, ranges.Range): 

293 mr.append( 

294 pg8000_Range(v.lower, v.upper, v.bounds, v.empty) 

295 ) 

296 else: 

297 mr.append(v) 

298 return mr 

299 else: 

300 return value 

301 

302 return to_multirange 

303 

304 def result_processor(self, dialect, coltype): 

305 def to_multirange(value): 

306 if value is None: 

307 return None 

308 else: 

309 return ranges.MultiRange( 

310 ranges.Range( 

311 v.lower, v.upper, bounds=v.bounds, empty=v.is_empty 

312 ) 

313 for v in value 

314 ) 

315 

316 return to_multirange 

317 

318 

319_server_side_id = util.counter() 

320 

321 

322class PGExecutionContext_pg8000(PGExecutionContext): 

323 def create_server_side_cursor(self): 

324 ident = "c_%s_%s" % (hex(id(self))[2:], hex(_server_side_id())[2:]) 

325 return ServerSideCursor(self._dbapi_connection.cursor(), ident) 

326 

327 def pre_exec(self): 

328 if not self.compiled: 

329 return 

330 

331 

332class ServerSideCursor: 

333 server_side = True 

334 

335 def __init__(self, cursor, ident): 

336 self.ident = ident 

337 self.cursor = cursor 

338 

339 @property 

340 def connection(self): 

341 return self.cursor.connection 

342 

343 @property 

344 def rowcount(self): 

345 return self.cursor.rowcount 

346 

347 @property 

348 def description(self): 

349 return self.cursor.description 

350 

351 def execute(self, operation, args=(), stream=None): 

352 op = "DECLARE " + self.ident + " NO SCROLL CURSOR FOR " + operation 

353 self.cursor.execute(op, args, stream=stream) 

354 return self 

355 

356 def executemany(self, operation, param_sets): 

357 self.cursor.executemany(operation, param_sets) 

358 return self 

359 

360 def fetchone(self): 

361 self.cursor.execute("FETCH FORWARD 1 FROM " + self.ident) 

362 return self.cursor.fetchone() 

363 

364 def fetchmany(self, num=None): 

365 if num is None: 

366 return self.fetchall() 

367 else: 

368 self.cursor.execute( 

369 "FETCH FORWARD " + str(int(num)) + " FROM " + self.ident 

370 ) 

371 return self.cursor.fetchall() 

372 

373 def fetchall(self): 

374 self.cursor.execute("FETCH FORWARD ALL FROM " + self.ident) 

375 return self.cursor.fetchall() 

376 

377 def close(self): 

378 self.cursor.execute("CLOSE " + self.ident) 

379 self.cursor.close() 

380 

381 def setinputsizes(self, *sizes): 

382 self.cursor.setinputsizes(*sizes) 

383 

384 def setoutputsize(self, size, column=None): 

385 pass 

386 

387 

388class PGCompiler_pg8000(PGCompiler): 

389 def visit_mod_binary(self, binary, operator, **kw): 

390 return ( 

391 self.process(binary.left, **kw) 

392 + " %% " 

393 + self.process(binary.right, **kw) 

394 ) 

395 

396 

397class PGIdentifierPreparer_pg8000(PGIdentifierPreparer): 

398 def __init__(self, *args, **kwargs): 

399 PGIdentifierPreparer.__init__(self, *args, **kwargs) 

400 self._double_percents = False 

401 

402 

403class PGDialect_pg8000(PGDialect): 

404 driver = "pg8000" 

405 supports_statement_cache = True 

406 

407 minimum_dbapi_version = util.VersionInfo((1, 16, 6)) 

408 

409 supports_unicode_statements = True 

410 

411 supports_unicode_binds = True 

412 

413 default_paramstyle = "format" 

414 supports_sane_multi_rowcount = True 

415 execution_ctx_cls = PGExecutionContext_pg8000 

416 statement_compiler = PGCompiler_pg8000 

417 preparer = PGIdentifierPreparer_pg8000 

418 supports_server_side_cursors = True 

419 

420 supports_native_json_serialization = False 

421 supports_native_json_deserialization = True 

422 dialect_injects_custom_json_deserializer = True 

423 

424 render_bind_cast = True 

425 

426 # reversed as of pg8000 1.16.6. 1.16.5 and lower 

427 # are no longer compatible 

428 description_encoding = None 

429 # description_encoding = "use_encoding" 

430 

431 colspecs = util.update_copy( 

432 PGDialect.colspecs, 

433 { 

434 sqltypes.String: _PGString, 

435 sqltypes.Numeric: _PGNumericNoBind, 

436 sqltypes.Float: _PGFloat, 

437 sqltypes.JSON: _PGJSON, 

438 sqltypes.Boolean: _PGBoolean, 

439 sqltypes.NullType: _PGNullType, 

440 JSONB: _PGJSONB, 

441 CITEXT: CITEXT, 

442 sqltypes.JSON.JSONPathType: _PGJSONPathType, 

443 sqltypes.JSON.JSONIndexType: _PGJSONIndexType, 

444 sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType, 

445 sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType, 

446 sqltypes.Interval: _PGInterval, 

447 INTERVAL: _PGInterval, 

448 sqltypes.DateTime: _PGTimeStamp, 

449 sqltypes.DateTime: _PGTimeStamp, 

450 sqltypes.Date: _PGDate, 

451 sqltypes.Time: _PGTime, 

452 sqltypes.Integer: _PGInteger, 

453 sqltypes.SmallInteger: _PGSmallInteger, 

454 sqltypes.BigInteger: _PGBigInteger, 

455 sqltypes.Enum: _PGEnum, 

456 sqltypes.ARRAY: _PGARRAY, 

457 OIDVECTOR: _PGOIDVECTOR, 

458 ranges.INT4RANGE: _Pg8000Range, 

459 ranges.INT8RANGE: _Pg8000Range, 

460 ranges.NUMRANGE: _Pg8000Range, 

461 ranges.DATERANGE: _Pg8000Range, 

462 ranges.TSRANGE: _Pg8000Range, 

463 ranges.TSTZRANGE: _Pg8000Range, 

464 ranges.INT4MULTIRANGE: _Pg8000MultiRange, 

465 ranges.INT8MULTIRANGE: _Pg8000MultiRange, 

466 ranges.NUMMULTIRANGE: _Pg8000MultiRange, 

467 ranges.DATEMULTIRANGE: _Pg8000MultiRange, 

468 ranges.TSMULTIRANGE: _Pg8000MultiRange, 

469 ranges.TSTZMULTIRANGE: _Pg8000MultiRange, 

470 }, 

471 ) 

472 

473 def __init__(self, client_encoding=None, **kwargs): 

474 PGDialect.__init__(self, **kwargs) 

475 self.client_encoding = client_encoding 

476 

477 if self._native_inet_types: 

478 raise NotImplementedError( 

479 "The pg8000 dialect does not fully implement " 

480 "ipaddress type handling; INET is supported by default, " 

481 "CIDR is not" 

482 ) 

483 

484 def retrieve_dbapi_version(self, dbapi): 

485 return util.parse_version_string(getattr(dbapi, "__version__", None)) 

486 

487 @classmethod 

488 def import_dbapi(cls): 

489 return __import__("pg8000") 

490 

491 def create_connect_args(self, url): 

492 opts = url.translate_connect_args(username="user") 

493 if "port" in opts: 

494 opts["port"] = int(opts["port"]) 

495 opts.update(url.query) 

496 return ([], opts) 

497 

498 def is_disconnect(self, e, connection, cursor): 

499 if isinstance(e, self.dbapi.InterfaceError) and "network error" in str( 

500 e 

501 ): 

502 # new as of pg8000 1.19.0 for broken connections 

503 return True 

504 

505 # connection was closed normally 

506 return "connection is closed" in str(e) 

507 

508 def get_isolation_level_values(self, dbapi_connection): 

509 return ( 

510 "AUTOCOMMIT", 

511 "READ COMMITTED", 

512 "READ UNCOMMITTED", 

513 "REPEATABLE READ", 

514 "SERIALIZABLE", 

515 ) 

516 

517 def set_isolation_level(self, dbapi_connection, level): 

518 level = level.replace("_", " ") 

519 

520 if level == "AUTOCOMMIT": 

521 dbapi_connection.autocommit = True 

522 else: 

523 dbapi_connection.autocommit = False 

524 cursor = dbapi_connection.cursor() 

525 cursor.execute( 

526 "SET SESSION CHARACTERISTICS AS TRANSACTION " 

527 f"ISOLATION LEVEL {level}" 

528 ) 

529 cursor.execute("COMMIT") 

530 cursor.close() 

531 

532 def detect_autocommit_setting(self, dbapi_conn) -> bool: 

533 return bool(dbapi_conn.autocommit) 

534 

535 def set_readonly(self, connection, value): 

536 cursor = connection.cursor() 

537 try: 

538 cursor.execute( 

539 "SET SESSION CHARACTERISTICS AS TRANSACTION %s" 

540 % ("READ ONLY" if value else "READ WRITE") 

541 ) 

542 cursor.execute("COMMIT") 

543 finally: 

544 cursor.close() 

545 

546 def get_readonly(self, connection): 

547 cursor = connection.cursor() 

548 try: 

549 cursor.execute("show transaction_read_only") 

550 val = cursor.fetchone()[0] 

551 finally: 

552 cursor.close() 

553 

554 return val == "on" 

555 

556 def set_deferrable(self, connection, value): 

557 cursor = connection.cursor() 

558 try: 

559 cursor.execute( 

560 "SET SESSION CHARACTERISTICS AS TRANSACTION %s" 

561 % ("DEFERRABLE" if value else "NOT DEFERRABLE") 

562 ) 

563 cursor.execute("COMMIT") 

564 finally: 

565 cursor.close() 

566 

567 def get_deferrable(self, connection): 

568 cursor = connection.cursor() 

569 try: 

570 cursor.execute("show transaction_deferrable") 

571 val = cursor.fetchone()[0] 

572 finally: 

573 cursor.close() 

574 

575 return val == "on" 

576 

577 def _set_client_encoding(self, dbapi_connection, client_encoding): 

578 cursor = dbapi_connection.cursor() 

579 cursor.execute(f""" 

580 SET CLIENT_ENCODING TO '{client_encoding.replace("'", "''")}'""") 

581 cursor.execute("COMMIT") 

582 cursor.close() 

583 

584 def do_begin_twophase(self, connection, xid): 

585 connection.connection.tpc_begin((0, xid, "")) 

586 

587 def do_prepare_twophase(self, connection, xid): 

588 connection.connection.tpc_prepare() 

589 

590 def do_rollback_twophase( 

591 self, connection, xid, is_prepared=True, recover=False 

592 ): 

593 connection.connection.tpc_rollback((0, xid, "")) 

594 

595 def do_commit_twophase( 

596 self, connection, xid, is_prepared=True, recover=False 

597 ): 

598 connection.connection.tpc_commit((0, xid, "")) 

599 

600 def do_recover_twophase(self, connection): 

601 return [row[1] for row in connection.connection.tpc_recover()] 

602 

603 def on_connect(self): 

604 fns = [] 

605 

606 def on_connect(conn): 

607 conn.py_types[quoted_name] = conn.py_types[str] 

608 

609 fns.append(on_connect) 

610 

611 if self.client_encoding is not None: 

612 

613 def on_connect(conn): 

614 self._set_client_encoding(conn, self.client_encoding) 

615 

616 fns.append(on_connect) 

617 

618 if self._native_inet_types is False: 

619 

620 def on_connect(conn): 

621 # inet 

622 conn.register_in_adapter(869, lambda s: s) 

623 

624 # cidr 

625 conn.register_in_adapter(650, lambda s: s) 

626 

627 fns.append(on_connect) 

628 

629 if self._json_deserializer: 

630 

631 def on_connect(conn): 

632 # json 

633 conn.register_in_adapter(114, self._json_deserializer) 

634 

635 # jsonb 

636 conn.register_in_adapter(3802, self._json_deserializer) 

637 

638 fns.append(on_connect) 

639 

640 if len(fns) > 0: 

641 

642 def on_connect(conn): 

643 for fn in fns: 

644 fn(conn) 

645 

646 return on_connect 

647 else: 

648 return None 

649 

650 @util.memoized_property 

651 def _dialect_specific_select_one(self): 

652 return ";" 

653 

654 

655dialect = PGDialect_pg8000