Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/psycopg2/extensions.py: 67%
85 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1"""psycopg extensions to the DBAPI-2.0
3This module holds all the extensions to the DBAPI-2.0 provided by psycopg.
5- `connection` -- the new-type inheritable connection class
6- `cursor` -- the new-type inheritable cursor class
7- `lobject` -- the new-type inheritable large object class
8- `adapt()` -- exposes the PEP-246_ compatible adapting mechanism used
9 by psycopg to adapt Python types to PostgreSQL ones
11.. _PEP-246: https://www.python.org/dev/peps/pep-0246/
12"""
13# psycopg/extensions.py - DBAPI-2.0 extensions specific to psycopg
14#
15# Copyright (C) 2003-2019 Federico Di Gregorio <fog@debian.org>
16# Copyright (C) 2020-2021 The Psycopg Team
17#
18# psycopg2 is free software: you can redistribute it and/or modify it
19# under the terms of the GNU Lesser General Public License as published
20# by the Free Software Foundation, either version 3 of the License, or
21# (at your option) any later version.
22#
23# In addition, as a special exception, the copyright holders give
24# permission to link this program with the OpenSSL library (or with
25# modified versions of OpenSSL that use the same license as OpenSSL),
26# and distribute linked combinations including the two.
27#
28# You must obey the GNU Lesser General Public License in all respects for
29# all of the code used other than OpenSSL.
30#
31# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
32# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
33# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
34# License for more details.
36import re as _re
38from psycopg2._psycopg import ( # noqa
39 BINARYARRAY, BOOLEAN, BOOLEANARRAY, BYTES, BYTESARRAY, DATE, DATEARRAY,
40 DATETIMEARRAY, DECIMAL, DECIMALARRAY, FLOAT, FLOATARRAY, INTEGER,
41 INTEGERARRAY, INTERVAL, INTERVALARRAY, LONGINTEGER, LONGINTEGERARRAY,
42 ROWIDARRAY, STRINGARRAY, TIME, TIMEARRAY, UNICODE, UNICODEARRAY,
43 AsIs, Binary, Boolean, Float, Int, QuotedString, )
45from psycopg2._psycopg import ( # noqa
46 PYDATE, PYDATETIME, PYDATETIMETZ, PYINTERVAL, PYTIME, PYDATEARRAY,
47 PYDATETIMEARRAY, PYDATETIMETZARRAY, PYINTERVALARRAY, PYTIMEARRAY,
48 DateFromPy, TimeFromPy, TimestampFromPy, IntervalFromPy, )
50from psycopg2._psycopg import ( # noqa
51 adapt, adapters, encodings, connection, cursor,
52 lobject, Xid, libpq_version, parse_dsn, quote_ident,
53 string_types, binary_types, new_type, new_array_type, register_type,
54 ISQLQuote, Notify, Diagnostics, Column, ConnectionInfo,
55 QueryCanceledError, TransactionRollbackError,
56 set_wait_callback, get_wait_callback, encrypt_password, )
59"""Isolation level values."""
60ISOLATION_LEVEL_AUTOCOMMIT = 0
61ISOLATION_LEVEL_READ_UNCOMMITTED = 4
62ISOLATION_LEVEL_READ_COMMITTED = 1
63ISOLATION_LEVEL_REPEATABLE_READ = 2
64ISOLATION_LEVEL_SERIALIZABLE = 3
65ISOLATION_LEVEL_DEFAULT = None
68"""psycopg connection status values."""
69STATUS_SETUP = 0
70STATUS_READY = 1
71STATUS_BEGIN = 2
72STATUS_SYNC = 3 # currently unused
73STATUS_ASYNC = 4 # currently unused
74STATUS_PREPARED = 5
76# This is a useful mnemonic to check if the connection is in a transaction
77STATUS_IN_TRANSACTION = STATUS_BEGIN
80"""psycopg asynchronous connection polling values"""
81POLL_OK = 0
82POLL_READ = 1
83POLL_WRITE = 2
84POLL_ERROR = 3
87"""Backend transaction status values."""
88TRANSACTION_STATUS_IDLE = 0
89TRANSACTION_STATUS_ACTIVE = 1
90TRANSACTION_STATUS_INTRANS = 2
91TRANSACTION_STATUS_INERROR = 3
92TRANSACTION_STATUS_UNKNOWN = 4
95def register_adapter(typ, callable):
96 """Register 'callable' as an ISQLQuote adapter for type 'typ'."""
97 adapters[(typ, ISQLQuote)] = callable
100# The SQL_IN class is the official adapter for tuples starting from 2.0.6.
101class SQL_IN:
102 """Adapt any iterable to an SQL quotable object."""
103 def __init__(self, seq):
104 self._seq = seq
105 self._conn = None
107 def prepare(self, conn):
108 self._conn = conn
110 def getquoted(self):
111 # this is the important line: note how every object in the
112 # list is adapted and then how getquoted() is called on it
113 pobjs = [adapt(o) for o in self._seq]
114 if self._conn is not None:
115 for obj in pobjs:
116 if hasattr(obj, 'prepare'):
117 obj.prepare(self._conn)
118 qobjs = [o.getquoted() for o in pobjs]
119 return b'(' + b', '.join(qobjs) + b')'
121 def __str__(self):
122 return str(self.getquoted())
125class NoneAdapter:
126 """Adapt None to NULL.
128 This adapter is not used normally as a fast path in mogrify uses NULL,
129 but it makes easier to adapt composite types.
130 """
131 def __init__(self, obj):
132 pass
134 def getquoted(self, _null=b"NULL"):
135 return _null
138def make_dsn(dsn=None, **kwargs):
139 """Convert a set of keywords into a connection strings."""
140 if dsn is None and not kwargs: 140 ↛ 141line 140 didn't jump to line 141, because the condition on line 140 was never true
141 return ''
143 # If no kwarg is specified don't mung the dsn, but verify it
144 if not kwargs: 144 ↛ 145line 144 didn't jump to line 145, because the condition on line 144 was never true
145 parse_dsn(dsn)
146 return dsn
148 # Override the dsn with the parameters
149 if 'database' in kwargs:
150 if 'dbname' in kwargs: 150 ↛ 151line 150 didn't jump to line 151, because the condition on line 150 was never true
151 raise TypeError(
152 "you can't specify both 'database' and 'dbname' arguments")
153 kwargs['dbname'] = kwargs.pop('database')
155 # Drop the None arguments
156 kwargs = {k: v for (k, v) in kwargs.items() if v is not None}
158 if dsn is not None: 158 ↛ 159line 158 didn't jump to line 159, because the condition on line 158 was never true
159 tmp = parse_dsn(dsn)
160 tmp.update(kwargs)
161 kwargs = tmp
163 dsn = " ".join(["{}={}".format(k, _param_escape(str(v)))
164 for (k, v) in kwargs.items()])
166 # verify that the returned dsn is valid
167 parse_dsn(dsn)
169 return dsn
172def _param_escape(s,
173 re_escape=_re.compile(r"([\\'])"),
174 re_space=_re.compile(r'\s')):
175 """
176 Apply the escaping rule required by PQconnectdb
177 """
178 if not s: 178 ↛ 179line 178 didn't jump to line 179, because the condition on line 178 was never true
179 return "''"
181 s = re_escape.sub(r'\\\1', s)
182 if re_space.search(s): 182 ↛ 183line 182 didn't jump to line 183, because the condition on line 182 was never true
183 s = "'" + s + "'"
185 return s
188# Create default json typecasters for PostgreSQL 9.2 oids
189from psycopg2._json import register_default_json, register_default_jsonb # noqa
191try:
192 JSON, JSONARRAY = register_default_json()
193 JSONB, JSONBARRAY = register_default_jsonb()
194except ImportError:
195 pass
197del register_default_json, register_default_jsonb
200# Create default Range typecasters
201from psycopg2. _range import Range # noqa
202del Range
205# Add the "cleaned" version of the encodings to the key.
206# When the encoding is set its name is cleaned up from - and _ and turned
207# uppercase, so an encoding not respecting these rules wouldn't be found in the
208# encodings keys and would raise an exception with the unicode typecaster
209for k, v in list(encodings.items()):
210 k = k.replace('_', '').replace('-', '').upper()
211 encodings[k] = v
213del k, v