Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def handle_ERROR_RESPONSE(self, data, ps):
responses = tuple(
(s[0:1], s[1:].decode(self._client_encoding)) for s in
data.split(NULL_BYTE))
msg_dict = dict(responses)
if msg_dict[RESPONSE_CODE] == "28000":
self.error = InterfaceError("md5 password authentication failed")
else:
self.error = ProgrammingError(*tuple(v for k, v in responses))
def __next__(self):
try:
return self._cached_rows.popleft()
except IndexError:
if self.ps is None:
raise ProgrammingError("A query hasn't been issued.")
elif len(self.ps['row_desc']) == 0:
raise ProgrammingError("no result set")
else:
raise StopIteration()
: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")
STRING = 1043
"""String type oid."""
NUMBER = 1700
"""Numeric type oid"""
DATETIME = 1114
"""Timestamp type oid"""
ROWID = 26
"""ROWID type oid"""
__all__ = [
Warning, Bytea, DataError, DatabaseError, connect, InterfaceError,
ProgrammingError, Error, OperationalError, IntegrityError, InternalError,
NotSupportedError, ArrayContentNotHomogenousError,
ArrayDimensionsNotConsistentError, ArrayContentNotSupportedError, utc,
Connection, Cursor, Binary, Date, DateFromTicks, Time, TimeFromTicks,
Timestamp, TimestampFromTicks, BINARY, Interval, PGEnum, PGJson, PGJsonb,
PGTsvector, PGText, PGVarchar]
"""Version string for pg8000.
ProgrammingError = property(lambda self: self._getError(ProgrammingError))
NotSupportedError = property(
"""
Raised when attempting to transmit an array where the base type is not
supported for binary data transfer by the interface.
"""
pass
class ArrayContentNotHomogenousError(ProgrammingError):
"""
Raised when attempting to transmit an array that doesn't contain only a
single type of object.
"""
pass
class ArrayDimensionsNotConsistentError(ProgrammingError):
"""
Raised when attempting to transmit an array that has inconsistent
multi-dimension sizes.
"""
pass
class Bytea(binary_type):
"""Bytea is a str-derived class that is mapped to a PostgreSQL byte array.
This class is only used in Python 2, the built-in ``bytes`` type is used in
Python 3.
"""
pass
def Date(year, month, day):
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")
except AttributeError:
raise ProgrammingError("attempting to use unexecuted cursor")
def handle_EMPTY_QUERY_RESPONSE(self, data, ps):
self.error = ProgrammingError("query was empty")
ProgrammingError = property(lambda self: self._getError(ProgrammingError))
NotSupportedError = property(
def handle_EMPTY_QUERY_RESPONSE(self, data, ps):
self.error = ProgrammingError("query was empty")