from datetime import date as Date, datetime as Datetime, time as Time
from itertools import count, islice
from time import localtime
from warnings import warn
import pg8000
from pg8000.converters import (
BIGINT,
BOOLEAN,
BOOLEAN_ARRAY,
BYTES,
CHAR,
CHAR_ARRAY,
DATE,
FLOAT,
FLOAT_ARRAY,
INET,
INT2VECTOR,
INTEGER,
INTEGER_ARRAY,
INTERVAL,
JSON,
JSONB,
MACADDR,
NAME,
NAME_ARRAY,
NULLTYPE,
NUMERIC,
NUMERIC_ARRAY,
OID,
PGInterval,
STRING,
TEXT,
TEXT_ARRAY,
TIME,
TIMESTAMP,
TIMESTAMPTZ,
UNKNOWN,
UUID_TYPE,
VARCHAR,
VARCHAR_ARRAY,
XID,
)
from pg8000.core import CoreConnection
from pg8000.exceptions import DatabaseError, Error, InterfaceError
from ._version import get_versions
__version__ = get_versions()["version"]
del get_versions
# Copyright (c) 2007-2009, Mathieu Fenniak
# Copyright (c) The Contributors
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# * Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# * The name of the author may not be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
__author__ = "Mathieu Fenniak"
ROWID = OID
apilevel = "2.0"
"""The DBAPI level supported, currently "2.0".
This property is part of the `DBAPI 2.0 specification
`_.
"""
threadsafety = 1
"""Integer constant stating the level of thread safety the DBAPI interface
supports. This DBAPI module supports sharing of the module only. Connections
and cursors my not be shared between threads. This gives pg8000 a threadsafety
value of 1.
This property is part of the `DBAPI 2.0 specification
`_.
"""
paramstyle = "format"
BINARY = bytes
def PgDate(year, month, day):
"""Constuct an object holding a date value.
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.date`
"""
return Date(year, month, day)
def PgTime(hour, minute, second):
"""Construct an object holding a time value.
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.time`
"""
return Time(hour, minute, second)
def Timestamp(year, month, day, hour, minute, second):
"""Construct an object holding a timestamp value.
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.datetime`
"""
return Datetime(year, month, day, hour, minute, second)
def DateFromTicks(ticks):
"""Construct an object holding a date value from the given ticks value
(number of seconds since the epoch).
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.date`
"""
return Date(*localtime(ticks)[:3])
def TimeFromTicks(ticks):
"""Construct an objet holding a time value from the given ticks value
(number of seconds since the epoch).
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.time`
"""
return Time(*localtime(ticks)[3:6])
def TimestampFromTicks(ticks):
"""Construct an object holding a timestamp value from the given ticks value
(number of seconds since the epoch).
This function is part of the `DBAPI 2.0 specification
`_.
:rtype: :class:`datetime.datetime`
"""
return Timestamp(*localtime(ticks)[:6])
def Binary(value):
"""Construct an object holding binary data.
This function is part of the `DBAPI 2.0 specification
`_.
"""
return value
def connect(
user,
host="localhost",
database=None,
port=5432,
password=None,
source_address=None,
unix_sock=None,
ssl_context=None,
timeout=None,
tcp_keepalive=True,
application_name=None,
replication=None,
):
return Connection(
user,
host=host,
database=database,
port=port,
password=password,
source_address=source_address,
unix_sock=unix_sock,
ssl_context=ssl_context,
timeout=timeout,
tcp_keepalive=tcp_keepalive,
application_name=application_name,
replication=replication,
)
apilevel = "2.0"
"""The DBAPI level supported, currently "2.0".
This property is part of the `DBAPI 2.0 specification
`_.
"""
threadsafety = 1
"""Integer constant stating the level of thread safety the DBAPI interface
supports. This DBAPI module supports sharing of the module only. Connections
and cursors my not be shared between threads. This gives pg8000 a threadsafety
value of 1.
This property is part of the `DBAPI 2.0 specification
`_.
"""
paramstyle = "format"
def convert_paramstyle(style, query, args):
# I don't see any way to avoid scanning the query string char by char,
# so we might as well take that careful approach and create a
# state-based scanner. We'll use int variables for the state.
OUTSIDE = 0 # outside quoted string
INSIDE_SQ = 1 # inside single-quote string '...'
INSIDE_QI = 2 # inside quoted identifier "..."
INSIDE_ES = 3 # inside escaped single-quote string, E'...'
INSIDE_PN = 4 # inside parameter name eg. :name
INSIDE_CO = 5 # inside inline comment eg. --
in_quote_escape = False
in_param_escape = False
placeholders = []
output_query = []
param_idx = map(lambda x: "$" + str(x), count(1))
state = OUTSIDE
prev_c = None
for i, c in enumerate(query):
if i + 1 < len(query):
next_c = query[i + 1]
else:
next_c = None
if state == OUTSIDE:
if c == "'":
output_query.append(c)
if prev_c == "E":
state = INSIDE_ES
else:
state = INSIDE_SQ
elif c == '"':
output_query.append(c)
state = INSIDE_QI
elif c == "-":
output_query.append(c)
if prev_c == "-":
state = INSIDE_CO
elif style == "qmark" and c == "?":
output_query.append(next(param_idx))
elif (
style == "numeric" and c == ":" and next_c not in ":=" and prev_c != ":"
):
# Treat : as beginning of parameter name if and only
# if it's the only : around
# Needed to properly process type conversions
# i.e. sum(x)::float
output_query.append("$")
elif style == "named" and c == ":" and next_c not in ":=" and prev_c != ":":
# Same logic for : as in numeric parameters
state = INSIDE_PN
placeholders.append("")
elif style == "pyformat" and c == "%" and next_c == "(":
state = INSIDE_PN
placeholders.append("")
elif style in ("format", "pyformat") and c == "%":
style = "format"
if in_param_escape:
in_param_escape = False
output_query.append(c)
else:
if next_c == "%":
in_param_escape = True
elif next_c == "s":
state = INSIDE_PN
output_query.append(next(param_idx))
else:
raise InterfaceError(
"Only %s and %% are supported in the query."
)
else:
output_query.append(c)
elif state == INSIDE_SQ:
if c == "'":
if in_quote_escape:
in_quote_escape = False
else:
if next_c == "'":
in_quote_escape = True
else:
state = OUTSIDE
output_query.append(c)
elif state == INSIDE_QI:
if c == '"':
state = OUTSIDE
output_query.append(c)
elif state == INSIDE_ES:
if c == "'" and prev_c != "\\":
# check for escaped single-quote
state = OUTSIDE
output_query.append(c)
elif state == INSIDE_PN:
if style == "named":
placeholders[-1] += c
if next_c is None or (not next_c.isalnum() and next_c != "_"):
state = OUTSIDE
try:
pidx = placeholders.index(placeholders[-1], 0, -1)
output_query.append("$" + str(pidx + 1))
del placeholders[-1]
except ValueError:
output_query.append("$" + str(len(placeholders)))
elif style == "pyformat":
if prev_c == ")" and c == "s":
state = OUTSIDE
try:
pidx = placeholders.index(placeholders[-1], 0, -1)
output_query.append("$" + str(pidx + 1))
del placeholders[-1]
except ValueError:
output_query.append("$" + str(len(placeholders)))
elif c in "()":
pass
else:
placeholders[-1] += c
elif style == "format":
state = OUTSIDE
elif state == INSIDE_CO:
output_query.append(c)
if c == "\n":
state = OUTSIDE
prev_c = c
if style in ("numeric", "qmark", "format"):
vals = args
else:
vals = tuple(args[p] for p in placeholders)
return "".join(output_query), vals
class Cursor:
def __init__(self, connection):
self._c = connection
self.arraysize = 1
self._context = None
self._row_iter = None
self._input_oids = None
@property
def connection(self):
warn("DB-API extension cursor.connection used", stacklevel=3)
return self._c
@property
def rowcount(self):
context = self._context
if context is None:
return -1
return context.row_count
@property
def description(self):
context = self._context
if context is None:
return None
row_desc = context.columns
if row_desc is None:
return None
if len(row_desc) == 0:
return None
columns = []
for col in row_desc:
columns.append((col["name"], col["type_oid"], None, None, None, None, None))
return columns
##
# Executes a database operation. Parameters may be provided as a sequence
# or mapping and will be bound to variables in the operation.
#
# Stability: Part of the DBAPI 2.0 specification.
def execute(self, operation, args=(), stream=None):
"""Executes a database operation. Parameters may be provided as a
sequence, or as a mapping, depending upon the value of
:data:`pg8000.paramstyle`.
This method is part of the `DBAPI 2.0 specification
`_.
:param operation:
The SQL statement to execute.
:param args:
If :data:`paramstyle` is ``qmark``, ``numeric``, or ``format``,
this argument should be an array of parameters to bind into the
statement. If :data:`paramstyle` is ``named``, the argument should
be a dict mapping of parameters. If the :data:`paramstyle` is
``pyformat``, the argument value may be either an array or a
mapping.
:param stream: This is a pg8000 extension for use with the PostgreSQL
`COPY
`_
command. For a COPY FROM the parameter must be a readable file-like
object, and for COPY TO it must be writable.
.. versionadded:: 1.9.11
"""
try:
if not self._c.in_transaction and not self._c.autocommit:
self._c.execute_unnamed("begin transaction")
statement, vals = convert_paramstyle(paramstyle, operation, args)
self._context = self._c.execute_unnamed(
statement, vals=vals, input_oids=self._input_oids, stream=stream
)
self._row_iter = iter(self._context.rows)
self._input_oids = None
except AttributeError as e:
if self._c is None:
raise InterfaceError("Cursor closed")
elif self._c._sock is None:
raise InterfaceError("connection is closed")
else:
raise e
self.input_types = []
def executemany(self, operation, param_sets):
"""Prepare a database operation, and then execute it against all
parameter sequences or mappings provided.
This method is part of the `DBAPI 2.0 specification
`_.
:param operation:
The SQL statement to execute
:param parameter_sets:
A sequence of parameters to execute the statement with. The values
in the sequence should be sequences or mappings of parameters, the
same as the args argument of the :meth:`execute` method.
"""
rowcounts = []
input_oids = self._input_oids
for parameters in param_sets:
self._input_oids = input_oids
self.execute(operation, parameters)
rowcounts.append(self._context.row_count)
self._context.row_count = -1 if -1 in rowcounts else sum(rowcounts)
def callproc(self, procname, parameters=None):
args = [] if parameters is None else parameters
operation = "CALL " + procname + "(" + ", ".join(["%s" for _ in args]) + ")"
try:
statement, vals = convert_paramstyle("format", operation, args)
self._context = self._c.execute_unnamed(statement, vals=vals)
self._row_iter = iter(self._context.rows)
except AttributeError as e:
if self._c is None:
raise InterfaceError("Cursor closed")
elif self._c._sock is None:
raise InterfaceError("connection is closed")
else:
raise e
def fetchone(self):
"""Fetch the next row of a query result set.
This method is part of the `DBAPI 2.0 specification
`_.
:returns:
A row as a sequence of field values, or ``None`` if no more rows
are available.
"""
try:
return next(self)
except StopIteration:
return None
except TypeError:
raise ProgrammingError("attempting to use unexecuted cursor")
def __iter__(self):
"""A cursor object is iterable to retrieve the rows from a query.
This is a DBAPI 2.0 extension.
"""
return self
def __next__(self):
try:
return next(self._row_iter)
except AttributeError:
if self._context is None:
raise ProgrammingError("A query hasn't been issued.")
else:
raise
except StopIteration as e:
if self._context is None:
raise ProgrammingError("A query hasn't been issued.")
elif len(self._context.columns) == 0:
raise ProgrammingError("no result set")
else:
raise e
def fetchmany(self, num=None):
"""Fetches the next set of rows of a query result.
This method is part of the `DBAPI 2.0 specification
`_.
:param size:
The number of rows to fetch when called. If not provided, the
:attr:`arraysize` attribute value is used instead.
:returns:
A sequence, each entry of which is a sequence of field values
making up a row. If no more rows are available, an empty sequence
will be returned.
"""
try:
return tuple(islice(self, self.arraysize if num is None else num))
except TypeError:
raise ProgrammingError("attempting to use unexecuted cursor")
def fetchall(self):
"""Fetches all remaining rows of a query result.
This method is part of the `DBAPI 2.0 specification
`_.
:returns:
A sequence, each entry of which is a sequence of field values
making up a row.
"""
try:
return tuple(self)
except TypeError:
raise ProgrammingError("attempting to use unexecuted cursor")
def close(self):
"""Closes the cursor.
This method is part of the `DBAPI 2.0 specification
`_.
"""
self._c = None
def setinputsizes(self, *sizes):
"""This method is part of the `DBAPI 2.0 specification"""
oids = []
for size in sizes:
if isinstance(size, int):
oid = size
else:
try:
oid, _ = self._c.py_types[size]
except KeyError:
oid = pg8000.converters.UNKNOWN
oids.append(oid)
self._input_oids = oids
def setoutputsize(self, size, column=None):
"""This method is part of the `DBAPI 2.0 specification
`_, however, it is not
implemented by pg8000.
"""
pass
class Connection(CoreConnection):
# DBAPI Extension: supply exceptions as attributes on the connection
Warning = property(lambda self: self._getError(Warning))
Error = property(lambda self: self._getError(Error))
InterfaceError = property(lambda self: self._getError(InterfaceError))
DatabaseError = property(lambda self: self._getError(DatabaseError))
OperationalError = property(lambda self: self._getError(OperationalError))
IntegrityError = property(lambda self: self._getError(IntegrityError))
InternalError = property(lambda self: self._getError(InternalError))
ProgrammingError = property(lambda self: self._getError(ProgrammingError))
NotSupportedError = property(lambda self: self._getError(NotSupportedError))
def _getError(self, error):
warn("DB-API extension connection.%s used" % error.__name__, stacklevel=3)
return error
def cursor(self):
"""Creates a :class:`Cursor` object bound to this
connection.
This function is part of the `DBAPI 2.0 specification
`_.
"""
return Cursor(self)
def commit(self):
"""Commits the current database transaction.
This function is part of the `DBAPI 2.0 specification
`_.
"""
self.execute_unnamed("commit")
def rollback(self):
"""Rolls back the current database transaction.
This function is part of the `DBAPI 2.0 specification
`_.
"""
if not self.in_transaction:
return
self.execute_unnamed("rollback")
def xid(self, format_id, global_transaction_id, branch_qualifier):
"""Create a Transaction IDs (only global_transaction_id is used in pg)
format_id and branch_qualifier are not used in postgres
global_transaction_id may be any string identifier supported by
postgres returns a tuple
(format_id, global_transaction_id, branch_qualifier)"""
return (format_id, global_transaction_id, branch_qualifier)
def tpc_begin(self, xid):
"""Begins a TPC transaction with the given transaction ID xid.
This method should be called outside of a transaction (i.e. nothing may
have executed since the last .commit() or .rollback()).
Furthermore, it is an error to call .commit() or .rollback() within the
TPC transaction. A ProgrammingError is raised, if the application calls
.commit() or .rollback() during an active TPC transaction.
This function is part of the `DBAPI 2.0 specification
`_.
"""
self._xid = xid
if self.autocommit:
self.execute_unnamed("begin transaction")
def tpc_prepare(self):
"""Performs the first phase of a transaction started with .tpc_begin().
A ProgrammingError is be raised if this method is called outside of a
TPC transaction.
After calling .tpc_prepare(), no statements can be executed until
.tpc_commit() or .tpc_rollback() have been called.
This function is part of the `DBAPI 2.0 specification
`_.
"""
self.execute_unnamed("PREPARE TRANSACTION '%s';" % (self._xid[1],))
def tpc_commit(self, xid=None):
"""When called with no arguments, .tpc_commit() commits a TPC
transaction previously prepared with .tpc_prepare().
If .tpc_commit() is called prior to .tpc_prepare(), a single phase
commit is performed. A transaction manager may choose to do this if
only a single resource is participating in the global transaction.
When called with a transaction ID xid, the database commits the given
transaction. If an invalid transaction ID is provided, a
ProgrammingError will be raised. This form should be called outside of
a transaction, and is intended for use in recovery.
On return, the TPC transaction is ended.
This function is part of the `DBAPI 2.0 specification
`_.
"""
if xid is None:
xid = self._xid
if xid is None:
raise ProgrammingError("Cannot tpc_commit() without a TPC transaction!")
try:
previous_autocommit_mode = self.autocommit
self.autocommit = True
if xid in self.tpc_recover():
self.execute_unnamed("COMMIT PREPARED '%s';" % (xid[1],))
else:
# a single-phase commit
self.commit()
finally:
self.autocommit = previous_autocommit_mode
self._xid = None
def tpc_rollback(self, xid=None):
"""When called with no arguments, .tpc_rollback() rolls back a TPC
transaction. It may be called before or after .tpc_prepare().
When called with a transaction ID xid, it rolls back the given
transaction. If an invalid transaction ID is provided, a
ProgrammingError is raised. This form should be called outside of a
transaction, and is intended for use in recovery.
On return, the TPC transaction is ended.
This function is part of the `DBAPI 2.0 specification
`_.
"""
if xid is None:
xid = self._xid
if xid is None:
raise ProgrammingError(
"Cannot tpc_rollback() without a TPC prepared transaction!"
)
try:
previous_autocommit_mode = self.autocommit
self.autocommit = True
if xid in self.tpc_recover():
# a two-phase rollback
self.execute_unnamed("ROLLBACK PREPARED '%s';" % (xid[1],))
else:
# a single-phase rollback
self.rollback()
finally:
self.autocommit = previous_autocommit_mode
self._xid = None
def tpc_recover(self):
"""Returns a list of pending transaction IDs suitable for use with
.tpc_commit(xid) or .tpc_rollback(xid).
This function is part of the `DBAPI 2.0 specification
`_.
"""
try:
previous_autocommit_mode = self.autocommit
self.autocommit = True
curs = self.cursor()
curs.execute("select gid FROM pg_prepared_xacts")
return [self.xid(0, row[0], "") for row in curs.fetchall()]
finally:
self.autocommit = previous_autocommit_mode
class Warning(Exception):
"""Generic exception raised for important database warnings like data
truncations. This exception is not currently used by pg8000.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class DataError(DatabaseError):
"""Generic exception raised for errors that are due to problems with the
processed data. This exception is not currently raised by pg8000.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class OperationalError(DatabaseError):
"""
Generic exception raised for errors that are related to the database's
operation and not necessarily under the control of the programmer. This
exception is currently never raised by pg8000.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class IntegrityError(DatabaseError):
"""
Generic exception raised when the relational integrity of the database is
affected. This exception is not currently raised by pg8000.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class InternalError(DatabaseError):
"""Generic exception raised when the database encounters an internal error.
This is currently only raised when unexpected state occurs in the pg8000
interface itself, and is typically the result of a interface bug.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class ProgrammingError(DatabaseError):
"""Generic exception raised for programming errors. For example, this
exception is raised if more parameter fields are in a query string than
there are available parameters.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class NotSupportedError(DatabaseError):
"""Generic exception raised in case a method or database API was used which
is not supported by the database.
This exception is part of the `DBAPI 2.0 specification
`_.
"""
pass
class ArrayContentNotSupportedError(NotSupportedError):
"""
Raised when attempting to transmit an array where the base type is not
supported for binary data transfer by the interface.
"""
pass
__all__ = [
"BIGINT",
"BINARY",
"BOOLEAN",
"BOOLEAN_ARRAY",
"BYTES",
"Binary",
"CHAR",
"CHAR_ARRAY",
"Connection",
"Cursor",
"DATE",
"DataError",
"DatabaseError",
"Date",
"DateFromTicks",
"Error",
"FLOAT",
"FLOAT_ARRAY",
"INET",
"INT2VECTOR",
"INTEGER",
"INTEGER_ARRAY",
"INTERVAL",
"IntegrityError",
"InterfaceError",
"InternalError",
"JSON",
"JSONB",
"MACADDR",
"NAME",
"NAME_ARRAY",
"NULLTYPE",
"NUMERIC",
"NUMERIC_ARRAY",
"NotSupportedError",
"OID",
"OperationalError",
"PGInterval",
"ProgrammingError",
"ROWID",
"STRING",
"TEXT",
"TEXT_ARRAY",
"TIME",
"TIMESTAMP",
"TIMESTAMPTZ",
"Time",
"TimeFromTicks",
"Timestamp",
"TimestampFromTicks",
"UNKNOWN",
"UUID_TYPE",
"VARCHAR",
"VARCHAR_ARRAY",
"Warning",
"XID",
"connect",
]