# This file is part of python-sql. The COPYRIGHT file at the top level of

# this repository contains the full copyright notices and license terms.

import numbers

import string

import warnings

from collections import defaultdict

from itertools import chain

from threading import current_thread, local

__version__ = '1.8.2'

__all__ = [

'Flavor', 'Table', 'Values', 'Literal', 'Column', 'Grouping', 'Conflict',

'Matched', 'MatchedUpdate', 'MatchedDelete',

'NotMatched', 'NotMatchedInsert',

'Rollup', 'Cube', 'Excluded', 'Join', 'Asc', 'Desc', 'NullsFirst',

'NullsLast', 'format2numeric']

def _escape_identifier(name):

return '"%s"' % name.replace('"', '""')

def alias(i, letters=string.ascii_lowercase):

'''

Generate a unique alias based on integer

>>> [alias(n) for n in range(6)]

['a', 'b', 'c', 'd', 'e', 'f']

>>> [alias(n) for n in range(26, 30)]

['ba', 'bb', 'bc', 'bd']

>>> [alias(26**n) for n in range(5)]

['b', 'ba', 'baa', 'baaa', 'baaaa']

'''

s = ''

length = len(letters)

while True:

r = i % length

s = letters[r] + s

i //= length

if i == 0:

break

return s

class Flavor(object):

'''

Contains the flavor of SQL

Contains:

limitstyle - state the type of pagination

max_limit - limit to use if there is no limit but an offset

paramstyle - state the type of parameter marker formatting

ilike - support ilike extension

no_as - doesn't support AS keyword for column and table

no_boolean - doesn't support boolean type

null_ordering - support NULL ordering

function_mapping - dictionary with Function to replace

filter_ - support filter on aggregate functions

escape_empty - support empty escape

'''

def __init__(self, limitstyle='limit', max_limit=None, paramstyle='format',

ilike=False, no_as=False, no_boolean=False, null_ordering=True,

function_mapping=None, filter_=False, escape_empty=False):

if limitstyle not in {'fetch', 'limit', 'rownum'}:

raise ValueError("unsupported limitstyle: %r" % limitstyle)

self.limitstyle = limitstyle

if (max_limit is not None

and not isinstance(max_limit, numbers.Integral)):

raise ValueError("unsupported max_limit: %r" % max_limit)

self.max_limit = max_limit

if paramstyle not in {'format', 'qmark'}:

raise ValueError("unsupported paramstyle: %r" % paramstyle)

self.paramstyle = paramstyle

self.ilike = bool(ilike)

self.no_as = bool(no_as)

self.no_boolean = bool(no_boolean)

self.null_ordering = bool(null_ordering)

self.function_mapping = dict(function_mapping or {})

self.filter_ = bool(filter_)

self.escape_empty = bool(escape_empty)

@property

def param(self):

if self.paramstyle == 'format':

return '%s'

elif self.paramstyle == 'qmark':

return '?'

@staticmethod

def set(flavor):

'''Set this thread's flavor to flavor.'''

current_thread().__sql_flavor__ = flavor

@staticmethod

def get():

'''

Return this thread's flavor.

If this thread does not yet have a flavor, returns a new flavor and

sets this thread's flavor.

'''

try:

return current_thread().__sql_flavor__

except AttributeError:

flavor = Flavor()

current_thread().__sql_flavor__ = flavor

return flavor

class AliasManager(object):

'''

Context Manager for unique alias generation

'''

__slots__ = ()

local = local()

local.alias = None

local.nested = 0

local.exclude = None

def __init__(self, exclude=None):

if exclude:

if getattr(self.local, 'exclude', None) is None:

self.local.exclude = []

self.local.exclude.extend(exclude)

@classmethod

def __enter__(cls):

if getattr(cls.local, 'alias', None) is None:

cls.local.alias = defaultdict(cls.alias_factory)

cls.local.nested = 0

if getattr(cls.local, 'exclude', None) is None:

cls.local.exclude = []

cls.local.nested += 1

@classmethod

def __exit__(cls, type, value, traceback):

cls.local.nested -= 1

if not cls.local.nested:

cls.local.alias = None

cls.local.exclude = None

@classmethod

def get(cls, from_):

if getattr(cls.local, 'alias', None) is None:

return ''

if from_ in cls.local.exclude:

return ''

return cls.local.alias[id(from_)]

@classmethod

def contains(cls, from_):

if getattr(cls.local, 'alias', None) is None:

return False

if from_ in cls.local.exclude:

return False

return id(from_) in cls.local.alias

@classmethod

def set(cls, from_, alias):

assert cls.local.alias.get(from_) is None

cls.local.alias[id(from_)] = alias

@classmethod

def alias_factory(cls):

i = len(cls.local.alias)

return alias(i)

def format2numeric(query, params):

'''

Convert format paramstyle query to numeric paramstyle

>>> format2numeric('SELECT * FROM table WHERE col = %s', ('foo',))

('SELECT * FROM table WHERE col = :0', ('foo',))

>>> format2numeric('SELECT * FROM table WHERE col1 = %s AND col2 = %s',

... ('foo', 'bar'))

('SELECT * FROM table WHERE col1 = :0 AND col2 = :1', ('foo', 'bar'))

'''

return (query % tuple(':%i' % i for i, _ in enumerate(params)), params)

class Query(object):

__slots__ = ('__weakref__',)

@property

def params(self):

return ()

def __iter__(self):

yield str(self)

yield self.params

def __or__(self, other):

return Union(self, other)

def __and__(self, other):

return Intersect(self, other)

def __sub__(self, other):

return Except(self, other)

class WithQuery(Query):

__slots__ = ('_with',)

def __init__(self, **kwargs):

self._with = None

self.with_ = kwargs.pop('with_', None)

super(Query, self).__init__(**kwargs)

@property

def with_(self):

return self._with

@with_.setter

def with_(self, value):

if value is not None:

if isinstance(value, With):

value = [value]

if any(not isinstance(w, With) for w in value):

raise ValueError("invalid with: %r" % value)

self._with = value

def _with_str(self):

if not self.with_:

return ''

recursive = (' RECURSIVE' if any(w.recursive for w in self.with_)

else '')

with_ = ('WITH%s ' % recursive

+ ', '.join(w.statement() for w in self.with_)

+ ' ')

return with_

def _with_params(self):

if not self.with_:

return ()

params = []

for w in self.with_:

params.extend(w.statement_params())

return tuple(params)

class FromItem(object):

__slots__ = ('__weakref__',)

@property

def alias(self):

return AliasManager.get(self)

@property

def has_alias(self):

return AliasManager.contains(self)

def __getattr__(self, name):

if name.startswith('__'):

raise AttributeError

return Column(self, name)

def __add__(self, other):

if not isinstance(other, FromItem):

return NotImplemented

return From((self, other))

def select(self, *args, **kwargs):

return From((self,)).select(*args, **kwargs)

def join(self, right, type_='INNER', condition=None):

return Join(self, right, type_=type_, condition=condition)

def left_join(self, right, condition=None):

return self.join(right, type_='LEFT', condition=condition)

def left_outer_join(self, right, condition=None):

return self.join(right, type_='LEFT OUTER', condition=condition)

def right_join(self, right, condition=None):

return self.join(right, type_='RIGHT', condition=condition)

def right_outer_join(self, right, condition=None):

return self.join(right, type_='RIGHT OUTER', condition=condition)

def full_join(self, right, condition=None):

return self.join(right, type_='FULL', condition=condition)

def full_outer_join(self, right, condition=None):

return self.join(right, type_='FULL OUTER', condition=condition)

def cross_join(self, right, condition=None):

return self.join(right, type_='CROSS', condition=condition)

def lateral(self):

return Lateral(self)

class Lateral(FromItem):

__slots__ = ('_from_item',)

def __init__(self, from_item):

self._from_item = from_item

def __str__(self):

template = '%s'

if isinstance(self._from_item, Query):

template = '(%s)'

return 'LATERAL ' + template % self._from_item

def __getattr__(self, name):

return getattr(self._from_item, name)

class With(FromItem):

__slots__ = ('columns', 'query', 'recursive')

def __init__(self, *columns, **kwargs):

self.recursive = kwargs.pop('recursive', False)

self.columns = columns

self.query = kwargs.pop('query', None)

super(With, self).__init__(**kwargs)

def statement(self):

columns = (' (%s)' % ', '.join('"%s"' % c for c in self.columns)

if self.columns else '')

return '"%s"%s AS (%s)' % (self.alias, columns, self.query)

def statement_params(self):

return self.query.params

def __str__(self):

return '"%s"' % self.alias

@property

def params(self):

return tuple()

class SelectQuery(WithQuery):

__slots__ = ('_order_by', '_limit', '_offset')

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

self._order_by = None

self._limit = None

self._offset = None

self.order_by = kwargs.pop('order_by', None)

self.limit = kwargs.pop('limit', None)

self.offset = kwargs.pop('offset', None)

super(SelectQuery, self).__init__(*args, **kwargs)

@property

def order_by(self):

return self._order_by

@order_by.setter

def order_by(self, value):

if value is not None:

if isinstance(value, Expression):

value = [value]

if any(not isinstance(col, Expression) for col in value):

raise ValueError("invalid order by: %r" % value)

self._order_by = value

@property

def _order_by_str(self):

order_by = ''

if self.order_by:

order_by = ' ORDER BY ' + ', '.join(map(str, self.order_by))

return order_by

@property

def limit(self):

return self._limit

@limit.setter

def limit(self, value):

if value is not None:

if not isinstance(value, numbers.Integral):

raise ValueError("invalid limit: %r" % value)

self._limit = value

@property

def offset(self):

return self._offset

@offset.setter

def offset(self, value):

if value is not None:

if not isinstance(value, numbers.Integral):

raise ValueError("invalid offset: %r" % value)

self._offset = value

@property

def _limit_offset_str(self):

param = Flavor.get().param

if Flavor.get().limitstyle == 'limit':

offset = ''

if self.offset:

offset = ' OFFSET %s' % param

limit = ''

if self.limit is not None:

limit = ' LIMIT %s' % param

elif self.offset:

max_limit = Flavor.get().max_limit

if max_limit:

limit = ' LIMIT %s' % max_limit

return limit + offset

else:

offset = ''

if self.offset:

offset = ' OFFSET (%s) ROWS' % param

fetch = ''

if self.limit is not None:

fetch = ' FETCH FIRST (%s) ROWS ONLY' % param

return offset + fetch

@property

def _limit_offset_params(self):

p = []

if Flavor.get().limitstyle == 'limit':

if self.limit is not None:

p.append(self.limit)

if self.offset:

p.append(self.offset)

else:

if self.offset:

p.append(self.offset)

if self.limit is not None:

p.append(self.limit)

return tuple(p)

def as_(self, output_name):

return As(self, output_name)

class Select(FromItem, SelectQuery):

__slots__ = ('_columns', '_where', '_group_by', '_having', '_for_',

'from_', '_distinct', '_distinct_on', '_windows')

def __init__(self, columns, from_=None, where=None, group_by=None,

having=None, for_=None, distinct=False, distinct_on=None,

windows=None, **kwargs):

self._distinct = False

self._distinct_on = []

self._columns = None

self._where = None

self._group_by = None

self._having = None

self._for_ = None

self._windows = []

super(Select, self).__init__(**kwargs)

self.distinct = distinct

self.distinct_on = distinct_on

self.columns = columns

self.from_ = from_

self.where = where

self.group_by = group_by

self.having = having

self.for_ = for_

self.windows = windows

@property

def distinct(self):

return bool(self._distinct or self._distinct_on)

@distinct.setter

def distinct(self, value):

self._distinct = bool(value)

@property

def distinct_on(self):

return self._distinct_on

@distinct_on.setter

def distinct_on(self, value):

if value is not None:

if isinstance(value, Expression):

value = [value]

if any(not isinstance(col, Expression) for col in value):

raise ValueError("invalid distinct on: %r" % value)

self._distinct_on = value

@property

def columns(self):

return self._columns

@columns.setter

def columns(self, value):

if any(

not isinstance(col, (Expression, SelectQuery))

for col in value):

raise ValueError("invalid columns: %r" % value)

self._columns = tuple(value)

@property

def where(self):

return self._where

@where.setter

def where(self, value):

from sql.operators import And, Or

if value is not None:

if not isinstance(value, (Expression, And, Or)):

raise ValueError("invalid where: %r" % value)

self._where = value

@property

def group_by(self):

return self._group_by

@group_by.setter

def group_by(self, value):

if value is not None:

if isinstance(value, Expression):

value = [value]

if any(not isinstance(col, Expression) for col in value):

raise ValueError("invalid group by: %r" % value)

self._group_by = value

@property

def having(self):

return self._having

@having.setter

def having(self, value):

from sql.operators import And, Or

if value is not None:

if not isinstance(value, (Expression, And, Or)):

raise ValueError("invalid having: %r" % value)

self._having = value

@property

def for_(self):

return self._for_

@for_.setter

def for_(self, value):

if value is not None:

if isinstance(value, For):

value = [value]

if any(not isinstance(f, For) for f in value):

raise ValueError("invalid for: %r" % value)

self._for_ = value

@property

def windows(self):

from sql.aggregate import Aggregate

from sql.functions import WindowFunction

windows = set()

if self._windows is not None:

for window in self._windows:

windows.add(window)

yield window

for column in self.columns:

window_function = None

if isinstance(column, (WindowFunction, Aggregate)):

window_function = column

elif (isinstance(column, As)

and isinstance(column.expression,

(WindowFunction, Aggregate))):

window_function = column.expression

if (window_function and window_function.window

and window_function.window not in windows):

windows.add(window_function.window)

yield window_function.window

@windows.setter

def windows(self, value):

if value is not None:

if any(not isinstance(w, Window) for w in value):

raise ValueError("invalid windows: %r" % value)

self._windows = value

@staticmethod

def _format_column(column):

if isinstance(column, As):

if isinstance(column.expression, Select):

expression = '(%s)' % column.expression

else:

expression = column.expression

if Flavor.get().no_as:

return '%s %s' % (expression, column)

else:

return '%s AS %s' % (expression, column)

else:

if isinstance(column, Select):

return '(%s)' % column

else:

return str(column)

def _rownum(self, func):

aliases = [c.output_name if isinstance(c, As) else None

for c in self.columns]

def columns(table):

if aliases and all(aliases):

return [Column(table, alias) for alias in aliases]

else:

return [Column(table, '*')]

limitselect = self.select(*columns(self))

if self.limit is not None:

max_row = self.limit

if self.offset is not None:

max_row += self.offset

limitselect.where = _rownum <= max_row

if self.offset is not None:

rnum = _rownum.as_('rnum')

limitselect.columns += (rnum,)

offsetselect = limitselect.select(*columns(limitselect),

where=rnum > self.offset)

query = offsetselect

else:

query = limitselect

self.limit, limit = None, self.limit

self.offset, offset = None, self.offset

query.for_, self.for_ = self.for_, None

try:

value = func(query)

finally:

self.limit = limit

self.offset = offset

self.for_ = query.for_

return value

def __str__(self):

if (Flavor.get().limitstyle == 'rownum'

and (self.limit is not None or self.offset is not None)):

return self._rownum(str)

ordinals = {}

for expression in chain(

self.group_by or [],

self.order_by or []):

if not isinstance(expression, As):

continue

for i, column in enumerate(self.columns, start=1):

if not isinstance(column, As):

continue

if column.output_name != expression.output_name:

continue

if (str(column.expression) != str(expression.expression)

or column.params != expression.params):

raise ValueError("%r != %r" % (expression, column))

ordinals[column.output_name] = i

def str_or_ordinal(expression):

if isinstance(expression, As):

expression = ordinals.get(expression.output_name, expression)

return str(expression)

with AliasManager():

if self.from_ is not None:

from_ = ' FROM %s' % self.from_

else:

from_ = ''

# format window before expressions to set alias

window = ', '.join(

'"%s" AS (%s)' % (w.alias, w) for w in self.windows)

if window:

window = ' WINDOW ' + window

if self.distinct:

distinct = 'DISTINCT '

if self.distinct_on:

distinct += ('ON (%s) '

% ', '.join(map(str, self.distinct_on)))

else:

distinct = ''

if self.columns:

columns = ', '.join(map(self._format_column, self.columns))

else:

columns = '*'

where = ''

if self.where:

where = ' WHERE ' + str(self.where)

group_by = ''

if self.group_by:

group_by = ' GROUP BY ' + ', '.join(

map(str_or_ordinal, self.group_by))

having = ''

if self.having:

having = ' HAVING ' + str(self.having)

for_ = ''

if self.for_ is not None:

for_ = ' ' + ' '.join(map(str, self.for_))

return (self._with_str()

+ 'SELECT %s%s%s' % (distinct, columns, from_)

+ where + group_by + having + window + self._order_by_str

+ self._limit_offset_str + for_)

@property

def params(self):

if (Flavor.get().limitstyle == 'rownum'

and (self.limit is not None or self.offset is not None)):

return self._rownum(lambda q: q.params)

p = []

with AliasManager():

# Set alias to window function to skip their params

for window in self.windows:

window.alias

p.extend(self._with_params())

for column in chain(self.distinct_on or (), self.columns):

if isinstance(column, As):

p.extend(column.expression.params)

p.extend(column.params)

if self.from_ is not None:

p.extend(self.from_.params)

if self.where:

p.extend(self.where.params)

if self.group_by:

for expression in self.group_by:

p.extend(expression.params)

if self.having:

p.extend(self.having.params)

for window in self.windows:

p.extend(window.params)

if self.order_by:

for expression in self.order_by:

p.extend(expression.params)

p.extend(self._limit_offset_params)

return tuple(p)

class Insert(WithQuery):

__slots__ = ('_table', '_columns', '_values', '_on_conflict', '_returning')

def __init__(

self, table, columns=None, values=None, returning=None,

on_conflict=None, **kwargs):

self._table = None

self._columns = None

self._values = None

self._on_conflict = None

self._returning = None

self.table = table

self.columns = columns

self.values = values

self.on_conflict = on_conflict

self.returning = returning

super(Insert, self).__init__(**kwargs)

@property

def table(self):

return self._table

@table.setter

def table(self, value):

if not isinstance(value, Table):

raise ValueError("invalid table: %r" % value)

self._table = value

@property

def columns(self):

return self._columns

@columns.setter

def columns(self, value):

if value is not None:

if any(

not isinstance(col, Column) or col.table != self.table

for col in value):

raise ValueError("invalid columns: %r" % value)

self._columns = value

@property

def values(self):

return self._values

@values.setter

def values(self, value):

if value is not None:

if not isinstance(value, (list, Select)):

raise ValueError("invalid values: %r" % value)

if isinstance(value, list):

value = Values(value)

self._values = value

@property

def on_conflict(self):

return self._on_conflict

@on_conflict.setter

def on_conflict(self, value):

if value is not None:

if not isinstance(value, Conflict) or value.table != self.table:

raise ValueError("invalid on conflict: %r" % value)

self._on_conflict = value

@property

def returning(self):

return self._returning

@returning.setter

def returning(self, value):

if value is not None:

if not isinstance(value, list):

raise ValueError("invalid returning: %r" % value)

self._returning = value

@staticmethod

def _format(value, param=None):

if param is None:

param = Flavor.get().param

if isinstance(value, Expression):

return str(value)

elif isinstance(value, Select):

return '(%s)' % value

else:

return param

def __str__(self):

columns = ''

if self.columns:

assert all(col.table == self.table for col in self.columns)

# Get columns without alias

columns = ', '.join(c.column_name for c in self.columns)

columns = ' (' + columns + ')'

with AliasManager():

if isinstance(self.values, Query):

values = ' %s' % str(self.values)

# TODO manage DEFAULT

elif self.values is None:

values = ' DEFAULT VALUES'

on_conflict = ''

if self.on_conflict:

on_conflict = ' %s' % self.on_conflict

returning = ''

if self.returning:

returning = ' RETURNING ' + ', '.join(

map(self._format, self.returning))

if on_conflict or returning:

table = '%s AS "%s"' % (self.table, self.table.alias)

else:

table = str(self.table)

return (self._with_str()

+ 'INSERT INTO %s' % table

+ columns + values + on_conflict + returning)

@property

def params(self):

p = []

p.extend(self._with_params())

if isinstance(self.values, Query):

p.extend(self.values.params)

if self.on_conflict:

p.extend(self.on_conflict.params)

if self.returning:

for exp in self.returning:

p.extend(exp.params)

return tuple(p)

class Conflict(object):

__slots__ = (

'_table', '_indexed_columns', '_index_where', '_columns', '_values',

'_where')

def __init__(

self, table, indexed_columns=None, index_where=None,

columns=None, values=None, where=None):

self._table = None

self._indexed_columns = None

self._index_where = None

self._columns = None

self._values = None

self._where = None

self.table = table

self.indexed_columns = indexed_columns

self.index_where = index_where

self.columns = columns

self.values = values

self.where = where

@property

def table(self):

return self._table

@table.setter

def table(self, value):

if not isinstance(value, Table):

raise ValueError("invalid table: %r" % value)

self._table = value

@property

def indexed_columns(self):

return self._indexed_columns

@indexed_columns.setter

def indexed_columns(self, value):

if value is not None:

if any(

not isinstance(col, Column) or col.table != self.table

for col in value):

raise ValueError("invalid indexed columns: %r" % value)

self._indexed_columns = value

@property

def index_where(self):

return self._index_where

@index_where.setter

def index_where(self, value):

from sql.operators import And, Or

if value is not None:

if not isinstance(value, (Expression, And, Or)):

raise ValueError("invalid index where: %r" % value)

self._index_where = value

@property

def columns(self):

return self._columns

@columns.setter

def columns(self, value):

if value is not None:

if any(

not isinstance(col, Column) or col.table != self.table

for col in value):

raise ValueError("invalid columns: %r" % value)

self._columns = value

@property

def values(self):

return self._values

@values.setter

def values(self, value):

if value is not None:

if not isinstance(value, (list, Select)):

raise ValueError("invalid values: %r" % value)

if isinstance(value, list):

value = Values([value])

self._values = value

@property

def where(self):

return self._where

@where.setter

def where(self, value):

from sql.operators import And, Or

if value is not None:

if not isinstance(value, (Expression, And, Or)):

raise ValueError("invalid where: %r" % value)

self._where = value

def __str__(self):

indexed_columns = ''

if self.indexed_columns:

assert all(c.table == self.table for c in self.indexed_columns)

# Get columns without alias

indexed_columns = ', '.join(

c.column_name for c in self.indexed_columns)

indexed_columns = ' (' + indexed_columns + ')'

if self.index_where:

indexed_columns += ' WHERE ' + str(self.index_where)

else:

assert not self.index_where

do = ''

if not self.columns:

assert not self.values

assert not self.where

do = 'NOTHING'

else:

assert all(c.table == self.table for c in self.columns)

# Get columns without alias

do = ', '.join(c.column_name for c in self.columns)

# TODO manage DEFAULT

values = str(self.values)

if values.startswith('VALUES'):

values = values[len('VALUES'):]

else:

values = ' (' + values + ')'

if len(self.columns) == 1:

# PostgreSQL would require ROW expression

# with single column with parenthesis

do = 'UPDATE SET ' + do + ' =' + values

else:

do = 'UPDATE SET (' + do + ') =' + values

if self.where:

do += ' WHERE %s' % self.where

return 'ON CONFLICT' + indexed_columns + ' DO ' + do

@property

def params(self):

p = []

if self.index_where:

p.extend(self.index_where.params)

if self.values:

p.extend(self.values.params)

if self.where:

p.extend(self.where.params)

return p

class Update(Insert):