#!/usr/bin/python

#

# PyGreSQL - a Python interface for the PostgreSQL database.

#

# This file contains the classic pg module.

#

# Copyright (c) 2022 by the PyGreSQL Development Team

#

# The notification handler is based on pgnotify which is

# Copyright (c) 2001 Ng Pheng Siong. All rights reserved.

#

# Please see the LICENSE.TXT file for specific restrictions.

"""PyGreSQL classic interface.

This pg module implements some basic database management stuff.

It includes the _pg module and builds on it, providing the higher

level wrapper class named DB with additional functionality.

This is known as the "classic" ("old style") PyGreSQL interface.

For a DB-API 2 compliant interface use the newer pgdb module.

"""

from __future__ import print_function, division

try:

from _pg import *

except ImportError as e:

import os

libpq = 'libpq.'

if os.name == 'nt':

libpq += 'dll'

import sys

paths = [path for path in os.environ["PATH"].split(os.pathsep)

if os.path.exists(os.path.join(path, libpq))]

if sys.version_info >= (3, 8):

# see https://docs.python.org/3/whatsnew/3.8.html#ctypes

for path in paths:

with os.add_dll_directory(os.path.abspath(path)):

try:

from _pg import *

except ImportError:

pass

else:

e = None

break

if paths:

libpq = 'compatible ' + libpq

else:

libpq += 'so'

if e:

# note: we could use "raise from e" here in Python 3

raise ImportError(

"Cannot import shared library for PyGreSQL,\n"

"probably because no %s is installed.\n%s" % (libpq, e))

__version__ = version

__all__ = [

'DB', 'Adapter',

'NotificationHandler', 'Typecasts',

'Bytea', 'Hstore', 'Json', 'Literal',

'Error', 'Warning',

'DataError', 'DatabaseError',

'IntegrityError', 'InterfaceError', 'InternalError',

'InvalidResultError', 'MultipleResultsError',

'NoResultError', 'NotSupportedError',

'OperationalError', 'ProgrammingError',

'INV_READ', 'INV_WRITE',

'POLLING_OK', 'POLLING_FAILED', 'POLLING_READING', 'POLLING_WRITING',

'SEEK_CUR', 'SEEK_END', 'SEEK_SET',

'TRANS_ACTIVE', 'TRANS_IDLE', 'TRANS_INERROR',

'TRANS_INTRANS', 'TRANS_UNKNOWN',

'cast_array', 'cast_hstore', 'cast_record',

'connect', 'escape_bytea', 'escape_string', 'unescape_bytea',

'get_array', 'get_bool', 'get_bytea_escaped',

'get_datestyle', 'get_decimal', 'get_decimal_point',

'get_defbase', 'get_defhost', 'get_defopt', 'get_defport', 'get_defuser',

'get_jsondecode', 'get_typecast',

'set_array', 'set_bool', 'set_bytea_escaped',

'set_datestyle', 'set_decimal', 'set_decimal_point',

'set_defbase', 'set_defhost', 'set_defopt',

'set_defpasswd', 'set_defport', 'set_defuser',

'set_jsondecode', 'set_query_helpers', 'set_typecast',

'version', '__version__']

import select

import warnings

import weakref

from datetime import date, time, datetime, timedelta, tzinfo

from decimal import Decimal

from math import isnan, isinf

from collections import namedtuple, OrderedDict

from operator import itemgetter

from functools import partial

from re import compile as regex

from json import loads as jsondecode, dumps as jsonencode

from uuid import UUID

try:

# noinspection PyUnresolvedReferences

from typing import Dict, List, Union

has_typing = True

except ImportError: # Python < 3.5

has_typing = False

try: # noinspection PyUnresolvedReferences,PyUnboundLocalVariable

long

except NameError: # Python >= 3.0

long = int

try: # noinspection PyUnresolvedReferences,PyUnboundLocalVariable

unicode

except NameError: # Python >= 3.0

unicode = str

try: # noinspection PyUnresolvedReferences,PyUnboundLocalVariable

basestring

except NameError: # Python >= 3.0

basestring = (str, bytes)

try:

from functools import lru_cache

except ImportError: # Python < 3.2

from functools import update_wrapper

try: # noinspection PyCompatibility

from _thread import RLock

except ImportError:

class RLock: # for builds without threads

def __enter__(self):

pass

def __exit__(self, exctype, excinst, exctb):

pass

def lru_cache(maxsize=128):

"""Simplified functools.lru_cache decorator for one argument."""

def decorator(function):

sentinel = object()

cache = {}

get = cache.get

lock = RLock()

root = []

root_full = [root, False]

root[:] = [root, root, None, None]

if maxsize == 0:

def wrapper(arg):

res = function(arg)

return res

elif maxsize is None:

def wrapper(arg):

res = get(arg, sentinel)

if res is not sentinel:

return res

res = function(arg)

cache[arg] = res

return res

else:

def wrapper(arg):

with lock:

link = get(arg)

if link is not None:

root = root_full[0]

prv, nxt, _arg, res = link

prv[1] = nxt

nxt[0] = prv

last = root[0]

last[1] = root[0] = link

link[0] = last

link[1] = root

return res

res = function(arg)

with lock:

root, full = root_full

if arg in cache:

pass

elif full:

oldroot = root

oldroot[2] = arg

oldroot[3] = res

root = root_full[0] = oldroot[1]

oldarg = root[2]

oldres = root[3] # noqa F481 (keep reference)

root[2] = root[3] = None

del cache[oldarg]

cache[arg] = oldroot

else:

last = root[0]

link = [last, root, arg, res]

last[1] = root[0] = cache[arg] = link

if len(cache) >= maxsize:

root_full[1] = True

return res

wrapper.__wrapped__ = function

return update_wrapper(wrapper, function)

return decorator

# Auxiliary classes and functions that are independent of a DB connection:

try: # noinspection PyUnresolvedReferences

from inspect import signature

except ImportError: # Python < 3.3

from inspect import getargspec

def get_args(func):

return getargspec(func).args

else:

def get_args(func):

return list(signature(func).parameters)

try:

from datetime import timezone

except ImportError: # Python < 3.2

class timezone(tzinfo):

"""Simple timezone implementation."""

def __init__(self, offset, name=None):

self.offset = offset

if not name:

minutes = self.offset.days * 1440 + self.offset.seconds // 60

if minutes < 0:

hours, minutes = divmod(-minutes, 60)

hours = -hours

else:

hours, minutes = divmod(minutes, 60)

name = 'UTC%+03d:%02d' % (hours, minutes)

self.name = name

def utcoffset(self, dt):

return self.offset

def tzname(self, dt):

return self.name

def dst(self, dt):

return None

timezone.utc = timezone(timedelta(0), 'UTC')

_has_timezone = False

else:

_has_timezone = True

# time zones used in Postgres timestamptz output

_timezones = dict(CET='+0100', EET='+0200', EST='-0500',

GMT='+0000', HST='-1000', MET='+0100', MST='-0700',

UCT='+0000', UTC='+0000', WET='+0000')

def _timezone_as_offset(tz):

if tz.startswith(('+', '-')):

if len(tz) < 5:

return tz + '00'

return tz.replace(':', '')

return _timezones.get(tz, '+0000')

def _get_timezone(tz):

tz = _timezone_as_offset(tz)

minutes = 60 * int(tz[1:3]) + int(tz[3:5])

if tz[0] == '-':

minutes = -minutes

return timezone(timedelta(minutes=minutes), tz)

def _oid_key(table):

"""Build oid key from a table name."""

return 'oid(%s)' % table

class Bytea(bytes):

"""Wrapper class for marking Bytea values."""

class Hstore(dict):

"""Wrapper class for marking hstore values."""

_re_quote = regex('^[Nn][Uu][Ll][Ll]$|[ ,=>]')

@classmethod

def _quote(cls, s):

if s is None:

return 'NULL'

if not isinstance(s, basestring):

s = str(s)

if not s:

return '""'

s = s.replace('"', '\\"')

if cls._re_quote.search(s):

s = '"%s"' % s

return s

def __str__(self):

q = self._quote

return ','.join('%s=>%s' % (q(k), q(v)) for k, v in self.items())

class Json:

"""Wrapper class for marking Json values."""

def __init__(self, obj, encode=None):

self.obj = obj

self.encode = encode or jsonencode

def __str__(self):

obj = self.obj

if isinstance(obj, basestring):

return obj

return self.encode(obj)

class _SimpleTypes(dict):

"""Dictionary mapping pg_type names to simple type names.

The corresponding Python types and simple names are also mapped.

"""

_type_aliases = {

'bool': [bool],

'bytea': [Bytea],

'date': ['interval', 'time', 'timetz', 'timestamp', 'timestamptz',

'abstime', 'reltime', # these are very old

'datetime', 'timedelta', # these do not really exist

date, time, datetime, timedelta],

'float': ['float4', 'float8', float],

'int': ['cid', 'int2', 'int4', 'int8', 'oid', 'xid', int],

'hstore': [Hstore], 'json': ['jsonb', Json], 'uuid': [UUID],

'num': ['numeric', Decimal], 'money': [],

'text': ['bpchar', 'char', 'name', 'varchar',

bytes, unicode, basestring]

} # type: Dict[str, List[Union[str, type]]]

if long is not int: # Python 2 has a separate long type

_type_aliases['num'].append(long)

# noinspection PyMissingConstructor

def __init__(self):

"""Initialize type mapping."""

for typ, keys in self._type_aliases.items():

keys = [typ] + keys

for key in keys:

self[key] = typ

if isinstance(key, str):

self['_%s' % key] = '%s[]' % typ

elif has_typing and not isinstance(key, tuple):

self[List[key]] = '%s[]' % typ

@staticmethod

def __missing__(key):

"""Unmapped types are interpreted as text."""

return 'text'

def get_type_dict(self):

"""Get a plain dictionary of only the types."""

return dict((key, typ) for key, typ in self.items()

if not isinstance(key, (str, tuple)))

_simpletypes = _SimpleTypes()

_simple_type_dict = _simpletypes.get_type_dict()

def _quote_if_unqualified(param, name):

"""Quote parameter representing a qualified name.

Puts a quote_ident() call around the given parameter unless

the name contains a dot, in which case the name is ambiguous

(could be a qualified name or just a name with a dot in it)

and must be quoted manually by the caller.

"""

if isinstance(name, basestring) and '.' not in name:

return 'quote_ident(%s)' % (param,)

return param

class _ParameterList(list):

"""Helper class for building typed parameter lists."""

def add(self, value, typ=None):

"""Typecast value with known database type and build parameter list.

If this is a literal value, it will be returned as is. Otherwise, a

placeholder will be returned and the parameter list will be augmented.

"""

# noinspection PyUnresolvedReferences

value = self.adapt(value, typ)

if isinstance(value, Literal):

return value

self.append(value)

return '$%d' % len(self)

class Literal(str):

"""Wrapper class for marking literal SQL values."""

class AttrDict(OrderedDict):

"""Simple read-only ordered dictionary for storing attribute names."""

def __init__(self, *args, **kw):

self._read_only = False

OrderedDict.__init__(self, *args, **kw)

self._read_only = True

error = self._read_only_error

self.clear = self.update = error

self.pop = self.setdefault = self.popitem = error

def __setitem__(self, key, value):

if self._read_only:

self._read_only_error()

OrderedDict.__setitem__(self, key, value)

def __delitem__(self, key):

if self._read_only:

self._read_only_error()

OrderedDict.__delitem__(self, key)

@staticmethod

def _read_only_error(*args, **kw):

raise TypeError('This object is read-only')

class Adapter:

"""Class providing methods for adapting parameters to the database."""

_bool_true_values = frozenset('t true 1 y yes on'.split())

_date_literals = frozenset(

'current_date current_time'

' current_timestamp localtime localtimestamp'.split())

_re_array_quote = regex(r'[{},"\\\s]|^[Nn][Uu][Ll][Ll]$')

_re_record_quote = regex(r'[(,"\\]')

_re_array_escape = _re_record_escape = regex(r'(["\\])')

def __init__(self, db):

self.db = weakref.proxy(db)

@classmethod

def _adapt_bool(cls, v):

"""Adapt a boolean parameter."""

if isinstance(v, basestring):

if not v:

return None

v = v.lower() in cls._bool_true_values

return 't' if v else 'f'

@classmethod

def _adapt_date(cls, v):

"""Adapt a date parameter."""

if not v:

return None

if isinstance(v, basestring) and v.lower() in cls._date_literals:

return Literal(v)

return v

@staticmethod

def _adapt_num(v):

"""Adapt a numeric parameter."""

if not v and v != 0:

return None

return v

_adapt_int = _adapt_float = _adapt_money = _adapt_num

def _adapt_bytea(self, v):

"""Adapt a bytea parameter."""

return self.db.escape_bytea(v)

def _adapt_json(self, v):

"""Adapt a json parameter."""

if not v:

return None

if isinstance(v, basestring):

return v

if isinstance(v, Json):

return str(v)

return self.db.encode_json(v)

def _adapt_hstore(self, v):

"""Adapt a hstore parameter."""

if not v:

return None

if isinstance(v, basestring):

return v

if isinstance(v, Hstore):

return str(v)

if isinstance(v, dict):

return str(Hstore(v))

raise TypeError('Hstore parameter %s has wrong type' % v)

def _adapt_uuid(self, v):

"""Adapt a UUID parameter."""

if not v:

return None

if isinstance(v, basestring):

return v

return str(v)

@classmethod

def _adapt_text_array(cls, v):

"""Adapt a text type array parameter."""

if isinstance(v, list):

adapt = cls._adapt_text_array

return '{%s}' % ','.join(adapt(v) for v in v)

if v is None:

return 'null'

if not v:

return '""'

v = str(v)

if cls._re_array_quote.search(v):

v = '"%s"' % cls._re_array_escape.sub(r'\\\1', v)

return v

_adapt_date_array = _adapt_text_array

@classmethod

def _adapt_bool_array(cls, v):

"""Adapt a boolean array parameter."""

if isinstance(v, list):

adapt = cls._adapt_bool_array

return '{%s}' % ','.join(adapt(v) for v in v)

if v is None:

return 'null'

if isinstance(v, basestring):

if not v:

return 'null'

v = v.lower() in cls._bool_true_values

return 't' if v else 'f'

@classmethod

def _adapt_num_array(cls, v):

"""Adapt a numeric array parameter."""

if isinstance(v, list):

adapt = cls._adapt_num_array

return '{%s}' % ','.join(adapt(v) for v in v)

if not v and v != 0:

return 'null'

return str(v)

_adapt_int_array = _adapt_float_array = _adapt_money_array = \

_adapt_num_array

def _adapt_bytea_array(self, v):

"""Adapt a bytea array parameter."""

if isinstance(v, list):

return b'{' + b','.join(

self._adapt_bytea_array(v) for v in v) + b'}'

if v is None:

return b'null'

return self.db.escape_bytea(v).replace(b'\\', b'\\\\')

def _adapt_json_array(self, v):

"""Adapt a json array parameter."""

if isinstance(v, list):

adapt = self._adapt_json_array

return '{%s}' % ','.join(adapt(v) for v in v)

if not v:

return 'null'

if not isinstance(v, basestring):

v = self.db.encode_json(v)

if self._re_array_quote.search(v):

v = '"%s"' % self._re_array_escape.sub(r'\\\1', v)

return v

def _adapt_record(self, v, typ):

"""Adapt a record parameter with given type."""

typ = self.get_attnames(typ).values()

if len(typ) != len(v):

raise TypeError('Record parameter %s has wrong size' % v)

adapt = self.adapt

value = []

for v, t in zip(v, typ):

v = adapt(v, t)

if v is None:

v = ''

elif not v:

v = '""'

else:

if isinstance(v, bytes):

if str is not bytes:

v = v.decode('ascii')

else:

v = str(v)

if self._re_record_quote.search(v):

v = '"%s"' % self._re_record_escape.sub(r'\\\1', v)

value.append(v)

return '(%s)' % ','.join(value)

def adapt(self, value, typ=None):

"""Adapt a value with known database type."""

if value is not None and not isinstance(value, Literal):

if typ:

simple = self.get_simple_name(typ)

else:

typ = simple = self.guess_simple_type(value) or 'text'

pg_str = getattr(value, '__pg_str__', None)

if pg_str:

value = pg_str(typ)

if simple == 'text':

pass

elif simple == 'record':

if isinstance(value, tuple):

value = self._adapt_record(value, typ)

elif simple.endswith('[]'):

if isinstance(value, list):

adapt = getattr(self, '_adapt_%s_array' % simple[:-2])

value = adapt(value)

else:

adapt = getattr(self, '_adapt_%s' % simple)

value = adapt(value)

return value

@staticmethod

def simple_type(name):

"""Create a simple database type with given attribute names."""

typ = DbType(name)

typ.simple = name

return typ

@staticmethod

def get_simple_name(typ):

"""Get the simple name of a database type."""

if isinstance(typ, DbType):

# noinspection PyUnresolvedReferences

return typ.simple

return _simpletypes[typ]

@staticmethod

def get_attnames(typ):

"""Get the attribute names of a composite database type."""

if isinstance(typ, DbType):

return typ.attnames

return {}

@classmethod

def guess_simple_type(cls, value):

"""Try to guess which database type the given value has."""

# optimize for most frequent types

try:

return _simple_type_dict[type(value)]

except KeyError:

pass

if isinstance(value, basestring):

return 'text'

if isinstance(value, bool):

return 'bool'

if isinstance(value, (int, long)):

return 'int'

if isinstance(value, float):

return 'float'

if isinstance(value, Decimal):

return 'num'

if isinstance(value, (date, time, datetime, timedelta)):

return 'date'

if isinstance(value, Bytea):

return 'bytea'

if isinstance(value, Json):

return 'json'

if isinstance(value, Hstore):

return 'hstore'

if isinstance(value, UUID):

return 'uuid'

if isinstance(value, list):

return '%s[]' % (cls.guess_simple_base_type(value) or 'text',)

if isinstance(value, tuple):

simple_type = cls.simple_type

guess = cls.guess_simple_type

# noinspection PyUnusedLocal

def get_attnames(self):

return AttrDict((str(n + 1), simple_type(guess(v)))

for n, v in enumerate(value))

typ = simple_type('record')

typ._get_attnames = get_attnames

return typ

@classmethod

def guess_simple_base_type(cls, value):

"""Try to guess the base type of a given array."""

for v in value:

if isinstance(v, list):

typ = cls.guess_simple_base_type(v)

else:

typ = cls.guess_simple_type(v)

if typ:

return typ

def adapt_inline(self, value, nested=False):

"""Adapt a value that is put into the SQL and needs to be quoted."""

if value is None:

return 'NULL'

if isinstance(value, Literal):

return value

if isinstance(value, Bytea):

value = self.db.escape_bytea(value)

if bytes is not str: # Python >= 3.0

value = value.decode('ascii')

elif isinstance(value, (datetime, date, time, timedelta)):

value = str(value)

if isinstance(value, basestring):

value = self.db.escape_string(value)

return "'%s'" % value

if isinstance(value, bool):

return 'true' if value else 'false'

if isinstance(value, float):

if isinf(value):

return "'-Infinity'" if value < 0 else "'Infinity'"

if isnan(value):

return "'NaN'"

return value

if isinstance(value, (int, long, Decimal)):

return value

if isinstance(value, list):

q = self.adapt_inline

s = '[%s]' if nested else 'ARRAY[%s]'

return s % ','.join(str(q(v, nested=True)) for v in value)

if isinstance(value, tuple):

q = self.adapt_inline

return '(%s)' % ','.join(str(q(v)) for v in value)

if isinstance(value, Json):

value = self.db.escape_string(str(value))

return "'%s'::json" % value

if isinstance(value, Hstore):

value = self.db.escape_string(str(value))

return "'%s'::hstore" % value

pg_repr = getattr(value, '__pg_repr__', None)

if not pg_repr:

raise InterfaceError(

'Do not know how to adapt type %s' % type(value))

value = pg_repr()

if isinstance(value, (tuple, list)):

value = self.adapt_inline(value)

return value

def parameter_list(self):

"""Return a parameter list for parameters with known database types.

The list has an add(value, typ) method that will build up the

list and return either the literal value or a placeholder.

"""

params = _ParameterList()

params.adapt = self.adapt

return params

def format_query(self, command, values=None, types=None, inline=False):

"""Format a database query using the given values and types.

The optional types describe the values and must be passed as a list,

tuple or string (that will be split on whitespace) when values are

passed as a list or tuple, or as a dict if values are passed as a dict.

If inline is set to True, then parameters will be passed inline

together with the query string.

"""

if not values:

return command, []

if inline and types:

raise ValueError('Typed parameters must be sent separately')

params = self.parameter_list()

if isinstance(values, (list, tuple)):

if inline:

adapt = self.adapt_inline

literals = [adapt(value) for value in values]

else:

add = params.add

if types:

if isinstance(types, basestring):

types = types.split()

if (not isinstance(types, (list, tuple))

or len(types) != len(values)):

raise TypeError('The values and types do not match')

literals = [add(value, typ)

for value, typ in zip(values, types)]

else:

literals = [add(value) for value in values]

command %= tuple(literals)

elif isinstance(values, dict):

# we want to allow extra keys in the dictionary,

# so we first must find the values actually used in the command

used_values = {}

literals = dict.fromkeys(values, '')

for key in values:

del literals[key]

try:

command % literals

except KeyError:

used_values[key] = values[key]

literals[key] = ''

values = used_values

if inline:

adapt = self.adapt_inline

literals = {key: adapt(value)

for key, value in values.items()}

else:

add = params.add

if types:

if not isinstance(types, dict):

raise TypeError('The values and types do not match')

literals = {key: add(values[key], types.get(key))

for key in sorted(values)}

else:

literals = {key: add(values[key])

for key in sorted(values)}

command %= literals

else:

raise TypeError('The values must be passed as tuple, list or dict')

return command, params

def cast_bool(value):

"""Cast a boolean value."""

if not get_bool():

return value

return value[0] == 't'

def cast_json(value):

"""Cast a JSON value."""

cast = get_jsondecode()

if not cast:

return value

return cast(value)

def cast_num(value):

"""Cast a numeric value."""

return (get_decimal() or float)(value)

def cast_money(value):

"""Cast a money value."""

point = get_decimal_point()

if not point:

return value

if point != '.':

value = value.replace(point, '.')

value = value.replace('(', '-')

value = ''.join(c for c in value if c.isdigit() or c in '.-')

return (get_decimal() or float)(value)

def cast_int2vector(value):

"""Cast an int2vector value."""

return [int(v) for v in value.split()]

def cast_date(value, connection):

"""Cast a date value."""

# The output format depends on the server setting DateStyle. The default

# setting ISO and the setting for German are actually unambiguous. The

# order of days and months in the other two settings is however ambiguous,

# so at least here we need to consult the setting to properly parse values.

if value == '-infinity':

return date.min

if value == 'infinity':

return date.max

value = value.split()

if value[-1] == 'BC':

return date.min

value = value[0]

if len(value) > 10:

return date.max

fmt = connection.date_format()

return datetime.strptime(value, fmt).date()

def cast_time(value):

"""Cast a time value."""

fmt = '%H:%M:%S.%f' if len(value) > 8 else '%H:%M:%S'

return datetime.strptime(value, fmt).time()

_re_timezone = regex('(.*)([+-].*)')

def cast_timetz(value):

"""Cast a timetz value."""

tz = _re_timezone.match(value)

if tz:

value, tz = tz.groups()

else:

tz = '+0000'

fmt = '%H:%M:%S.%f' if len(value) > 8 else '%H:%M:%S'

if _has_timezone:

value += _timezone_as_offset(tz)

fmt += '%z'

return datetime.strptime(value, fmt).timetz()

return datetime.strptime(value, fmt).timetz().replace(

tzinfo=_get_timezone(tz))

def cast_timestamp(value, connection):

"""Cast a timestamp value."""

if value == '-infinity':

return datetime.min

if value == 'infinity':

return datetime.max

value = value.split()

if value[-1] == 'BC':

return datetime.min

fmt = connection.date_format()

if fmt.endswith('-%Y') and len(value) > 2:

value = value[1:5]

if len(value[3]) > 4:

return datetime.max

fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',

'%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']

else:

if len(value[0]) > 10:

return datetime.max

fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']

return datetime.strptime(' '.join(value), ' '.join(fmt))

def cast_timestamptz(value, connection):

"""Cast a timestamptz value."""

if value == '-infinity':

return datetime.min

if value == 'infinity':

return datetime.max

value = value.split()

if value[-1] == 'BC':

return datetime.min

fmt = connection.date_format()

if fmt.endswith('-%Y') and len(value) > 2:

value = value[1:]

if len(value[3]) > 4:

return datetime.max

fmt = ['%d %b' if fmt.startswith('%d') else '%b %d',

'%H:%M:%S.%f' if len(value[2]) > 8 else '%H:%M:%S', '%Y']

value, tz = value[:-1], value[-1]

else:

if fmt.startswith('%Y-'):

tz = _re_timezone.match(value[1])

if tz:

value[1], tz = tz.groups()

else:

tz = '+0000'

else:

value, tz = value[:-1], value[-1]

if len(value[0]) > 10:

return datetime.max

fmt = [fmt, '%H:%M:%S.%f' if len(value[1]) > 8 else '%H:%M:%S']

if _has_timezone:

value.append(_timezone_as_offset(tz))

fmt.append('%z')

return datetime.strptime(' '.join(value), ' '.join(fmt))

return datetime.strptime(' '.join(value), ' '.join(fmt)).replace(

tzinfo=_get_timezone(tz))

_re_interval_sql_standard = regex(

'(?:([+-])?([0-9]+)-([0-9]+) ?)?'

'(?:([+-]?[0-9]+)(?!:) ?)?'

'(?:([+-])?([0-9]+):([0-9]+):([0-9]+)(?:\\.([0-9]+))?)?')

_re_interval_postgres = regex(

'(?:([+-]?[0-9]+) ?years? ?)?'

'(?:([+-]?[0-9]+) ?mons? ?)?'

'(?:([+-]?[0-9]+) ?days? ?)?'

'(?:([+-])?([0-9]+):([0-9]+):([0-9]+)(?:\\.([0-9]+))?)?')

_re_interval_postgres_verbose = regex(

'@ ?(?:([+-]?[0-9]+) ?years? ?)?'

'(?:([+-]?[0-9]+) ?mons? ?)?'

'(?:([+-]?[0-9]+) ?days? ?)?'

'(?:([+-]?[0-9]+) ?hours? ?)?'

'(?:([+-]?[0-9]+) ?mins? ?)?'

'(?:([+-])?([0-9]+)(?:\\.([0-9]+))? ?secs?)? ?(ago)?')

_re_interval_iso_8601 = regex(

'P(?:([+-]?[0-9]+)Y)?'

'(?:([+-]?[0-9]+)M)?'

'(?:([+-]?[0-9]+)D)?'

'(?:T(?:([+-]?[0-9]+)H)?'

'(?:([+-]?[0-9]+)M)?'

'(?:([+-])?([0-9]+)(?:\\.([0-9]+))?S)?)?')

def cast_interval(value):

"""Cast an interval value."""

# The output format depends on the server setting IntervalStyle, but it's

# not necessary to consult this setting to parse it. It's faster to just

# check all possible formats, and there is no ambiguity here.

m = _re_interval_iso_8601.match(value)

if m: