[tryton-debian-vcs] python-sql branch upstream updated. upstream/0.4-1-ge0b8408
Mathias Behrle
tryton-debian-vcs at alioth.debian.org
Wed Feb 18 11:16:44 UTC 2015
The following commit has been merged in the upstream branch:
https://alioth.debian.org/plugins/scmgit/cgi-bin/gitweb.cgi/?p=tryton/python-sql.git;a=commitdiff;h=upstream/0.4-1-ge0b8408
commit e0b84083e1df870cbca3851f88ac2cdd5e829e77
Author: Mathias Behrle <mathiasb at m9s.biz>
Date: Wed Feb 18 12:13:30 2015 +0100
Adding upstream version 0.5.
Signed-off-by: Mathias Behrle <mathiasb at m9s.biz>
diff --git a/CHANGELOG b/CHANGELOG
index b3144e4..e11ec39 100644
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,3 +1,10 @@
+Version 0.5 - 2015-01-29
+* Add schema
+* Add Common Table Expression
+* Escape Mod operator '%' with format paramstyle
+* Deprecate Interesect in favor of Intersect
+* Add Values
+
Version 0.4 - 2014-08-03
* Fix Mapping in Trim function
* Add __truediv__ alias of __div__ for Python3
diff --git a/PKG-INFO b/PKG-INFO
index 85cba71..288675b 100644
--- a/PKG-INFO
+++ b/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: python-sql
-Version: 0.4
+Version: 0.5
Summary: Library to write SQL queries
Home-page: http://code.google.com/p/python-sql/
Author: B2CK
@@ -101,6 +101,12 @@ Description: python-sql
>>> tuple(subselect.select(subselect.user))
('SELECT "a"."user" FROM (SELECT "b"."user" FROM "user_group" AS "b" WHERE ("b"."active" = %s)) AS "a"', (True,))
+ Select on other schema::
+
+ >>> other_table = Table('user', 'myschema')
+ >>> tuple(other_table.select())
+ ('SELECT * FROM "myschema"."user" AS "a"', ())
+
Insert query with default values::
>>> tuple(user.insert())
diff --git a/README b/README
index 57f08ee..f7ee30f 100644
--- a/README
+++ b/README
@@ -93,6 +93,12 @@ Select with sub-select::
>>> tuple(subselect.select(subselect.user))
('SELECT "a"."user" FROM (SELECT "b"."user" FROM "user_group" AS "b" WHERE ("b"."active" = %s)) AS "a"', (True,))
+Select on other schema::
+
+ >>> other_table = Table('user', 'myschema')
+ >>> tuple(other_table.select())
+ ('SELECT * FROM "myschema"."user" AS "a"', ())
+
Insert query with default values::
>>> tuple(user.insert())
diff --git a/python_sql.egg-info/PKG-INFO b/python_sql.egg-info/PKG-INFO
index 85cba71..288675b 100644
--- a/python_sql.egg-info/PKG-INFO
+++ b/python_sql.egg-info/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: python-sql
-Version: 0.4
+Version: 0.5
Summary: Library to write SQL queries
Home-page: http://code.google.com/p/python-sql/
Author: B2CK
@@ -101,6 +101,12 @@ Description: python-sql
>>> tuple(subselect.select(subselect.user))
('SELECT "a"."user" FROM (SELECT "b"."user" FROM "user_group" AS "b" WHERE ("b"."active" = %s)) AS "a"', (True,))
+ Select on other schema::
+
+ >>> other_table = Table('user', 'myschema')
+ >>> tuple(other_table.select())
+ ('SELECT * FROM "myschema"."user" AS "a"', ())
+
Insert query with default values::
>>> tuple(user.insert())
diff --git a/python_sql.egg-info/SOURCES.txt b/python_sql.egg-info/SOURCES.txt
index 7c24f7a..66256d5 100644
--- a/python_sql.egg-info/SOURCES.txt
+++ b/python_sql.egg-info/SOURCES.txt
@@ -17,6 +17,7 @@ sql/tests/test_alias.py
sql/tests/test_as.py
sql/tests/test_cast.py
sql/tests/test_column.py
+sql/tests/test_combining_query.py
sql/tests/test_conditionals.py
sql/tests/test_delete.py
sql/tests/test_for.py
@@ -27,4 +28,7 @@ sql/tests/test_literal.py
sql/tests/test_operators.py
sql/tests/test_order.py
sql/tests/test_select.py
-sql/tests/test_update.py
\ No newline at end of file
+sql/tests/test_table.py
+sql/tests/test_update.py
+sql/tests/test_values.py
+sql/tests/test_with.py
\ No newline at end of file
diff --git a/sql/__init__.py b/sql/__init__.py
index fb09560..52fb5f7 100644
--- a/sql/__init__.py
+++ b/sql/__init__.py
@@ -1,8 +1,8 @@
# -*- coding: utf-8 -*-
#
-# Copyright (c) 2011-2013, Cédric Krier
-# Copyright (c) 2013, Nicolas Évrard
-# Copyright (c) 2011-2013, B2CK
+# Copyright (c) 2011-2015, Cédric Krier
+# Copyright (c) 2013-2014, Nicolas Évrard
+# Copyright (c) 2011-2015, B2CK
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
@@ -29,11 +29,12 @@
from __future__ import division
-__version__ = '0.4'
-__all__ = ['Flavor', 'Table', 'Literal', 'Column', 'Join', 'Asc', 'Desc']
+__version__ = '0.5'
+__all__ = ['Flavor', 'Table', 'Values', 'Literal', 'Column', 'Join',
+ 'Asc', 'Desc']
import string
-from functools import partial
+import warnings
from threading import local, currentThread
from collections import defaultdict
@@ -115,12 +116,21 @@ class AliasManager(object):
local = local()
local.alias = None
local.nested = 0
+ local.exclude = None
+
+ def __init__(self, exclude=None):
+ if exclude:
+ if self.local.exclude 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
@@ -128,17 +138,20 @@ class AliasManager(object):
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 ''
- return cls.local.alias[from_]
+ if from_ in cls.local.exclude:
+ return ''
+ return cls.local.alias[id(from_)]
@classmethod
def set(cls, from_, alias):
assert cls.local.alias.get(from_) is None
- cls.local.alias[from_] = alias
+ cls.local.alias[id(from_)] = alias
@classmethod
def alias_factory(cls):
@@ -157,6 +170,54 @@ class Query(object):
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]
+ assert all(isinstance(w, With) for w in 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__ = ()
@@ -166,6 +227,8 @@ class FromItem(object):
return AliasManager.get(self)
def __getattr__(self, name):
+ if name.startswith('__'):
+ raise AttributeError
return Column(self, name)
def __add__(self, other):
@@ -179,21 +242,46 @@ class FromItem(object):
return Join(self, right, type_=type_, condition=condition)
-class _SelectQueryMixin(object):
- __slots__ = ('__order_by', '__limit', '__offset')
+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 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 = 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(_SelectQueryMixin, self).__init__(*args, **kwargs)
+ super(SelectQuery, self).__init__(*args, **kwargs)
@property
def order_by(self):
- return self.__order_by
+ return self._order_by
@order_by.setter
def order_by(self, value):
@@ -201,7 +289,7 @@ class _SelectQueryMixin(object):
if isinstance(value, Expression):
value = [value]
assert all(isinstance(col, Expression) for col in value)
- self.__order_by = value
+ self._order_by = value
@property
def _order_by_str(self):
@@ -212,13 +300,13 @@ class _SelectQueryMixin(object):
@property
def limit(self):
- return self.__limit
+ return self._limit
@limit.setter
def limit(self, value):
if value is not None:
assert isinstance(value, (int, long))
- self.__limit = value
+ self._limit = value
@property
def _limit_str(self):
@@ -233,13 +321,13 @@ class _SelectQueryMixin(object):
@property
def offset(self):
- return self.__offset
+ return self._offset
@offset.setter
def offset(self, value):
if value is not None:
assert isinstance(value, (int, long))
- self.__offset = value
+ self._offset = value
@property
def _offset_str(self):
@@ -249,17 +337,17 @@ class _SelectQueryMixin(object):
return offset
-class Select(Query, FromItem, _SelectQueryMixin):
- __slots__ = ('__columns', '__where', '__group_by', '__having', '__for_',
+class Select(FromItem, SelectQuery):
+ __slots__ = ('_columns', '_where', '_group_by', '_having', '_for_',
'from_')
def __init__(self, columns, from_=None, where=None, group_by=None,
having=None, for_=None, **kwargs):
- self.__columns = None
- self.__where = None
- self.__group_by = None
- self.__having = None
- self.__for_ = None
+ self._columns = None
+ self._where = None
+ self._group_by = None
+ self._having = None
+ self._for_ = None
super(Select, self).__init__(**kwargs)
# TODO ALL|DISTINCT
self.columns = columns
@@ -271,27 +359,27 @@ class Select(Query, FromItem, _SelectQueryMixin):
@property
def columns(self):
- return self.__columns
+ return self._columns
@columns.setter
def columns(self, value):
assert all(isinstance(col, Expression) for col in value)
- self.__columns = tuple(value)
+ self._columns = tuple(value)
@property
def where(self):
- return self.__where
+ return self._where
@where.setter
def where(self, value):
from sql.operators import And, Or
if value is not None:
assert isinstance(value, (Expression, And, Or))
- self.__where = value
+ self._where = value
@property
def group_by(self):
- return self.__group_by
+ return self._group_by
@group_by.setter
def group_by(self, value):
@@ -299,22 +387,22 @@ class Select(Query, FromItem, _SelectQueryMixin):
if isinstance(value, Expression):
value = [value]
assert all(isinstance(col, Expression) for col in value)
- self.__group_by = value
+ self._group_by = value
@property
def having(self):
- return self.__having
+ return self._having
@having.setter
def having(self, value):
from sql.operators import And, Or
if value is not None:
assert isinstance(value, (Expression, And, Or))
- self.__having = value
+ self._having = value
@property
def for_(self):
- return self.__for_
+ return self._for_
@for_.setter
def for_(self, value):
@@ -322,7 +410,7 @@ class Select(Query, FromItem, _SelectQueryMixin):
if isinstance(value, For):
value = [value]
assert all(isinstance(f, For) for f in value)
- self.__for_ = value
+ self._for_ = value
@staticmethod
def _format_column(column):
@@ -350,13 +438,15 @@ class Select(Query, FromItem, _SelectQueryMixin):
for_ = ''
if self.for_ is not None:
for_ = ' ' + ' '.join(map(str, self.for_))
- return ('SELECT %s FROM %s' % (columns, from_) + where + group_by
- + having + self._order_by_str + self._limit_str +
- self._offset_str + for_)
+ return (self._with_str()
+ + 'SELECT %s FROM %s' % (columns, from_)
+ + where + group_by + having + self._order_by_str
+ + self._limit_str + self._offset_str + for_)
@property
def params(self):
p = []
+ p.extend(self._with_params())
for column in self.columns:
if isinstance(column, As):
p.extend(column.expression.params)
@@ -374,68 +464,63 @@ class Select(Query, FromItem, _SelectQueryMixin):
p.extend(expression.params)
return tuple(p)
- def __or__(self, other):
- return Union(self, other)
-
- def __and__(self, other):
- return Interesect(self, other)
-
- def __sub__(self, other):
- return Except(self, other)
-
-class Insert(Query):
- __slots__ = ('__table', '__columns', '__values', '__returning')
+class Insert(WithQuery):
+ __slots__ = ('_table', '_columns', '_values', '_returning')
- def __init__(self, table, columns=None, values=None, returning=None):
- self.__table = None
- self.__columns = None
- self.__values = None
- self.__returning = None
+ def __init__(self, table, columns=None, values=None, returning=None,
+ **kwargs):
+ self._table = None
+ self._columns = None
+ self._values = None
+ self._returning = None
self.table = table
self.columns = columns
self.values = values
self.returning = returning
+ super(Insert, self).__init__(**kwargs)
@property
def table(self):
- return self.__table
+ return self._table
@table.setter
def table(self, value):
assert isinstance(value, Table)
- self.__table = value
+ self._table = value
@property
def columns(self):
- return self.__columns
+ return self._columns
@columns.setter
def columns(self, value):
if value is not None:
assert all(isinstance(col, Column) for col in value)
assert all(col.table == self.table for col in value)
- self.__columns = value
+ self._columns = value
@property
def values(self):
- return self.__values
+ return self._values
@values.setter
def values(self, value):
if value is not None:
assert isinstance(value, (list, Select))
- self.__values = value
+ if isinstance(value, list):
+ value = Values(value)
+ self._values = value
@property
def returning(self):
- return self.__returning
+ return self._returning
@returning.setter
def returning(self, value):
if value is not None:
assert isinstance(value, list)
- self.__returning = value
+ self._returning = value
@staticmethod
def _format(value, param=None):
@@ -453,32 +538,24 @@ class Insert(Query):
if self.columns:
assert all(col.table == self.table for col in self.columns)
columns = ' (' + ', '.join(map(str, self.columns)) + ')'
- if isinstance(self.values, list):
- format_ = partial(self._format, param=Flavor.get().param)
- values = ' VALUES ' + ', '.join(
- '(' + ', '.join(map(format_, v)) + ')'
- for v in self.values)
- # TODO manage DEFAULT
- elif isinstance(self.values, Select):
+ if isinstance(self.values, Query):
values = ' %s' % str(self.values)
+ # TODO manage DEFAULT
elif self.values is None:
values = ' DEFAULT VALUES'
returning = ''
if self.returning:
returning = ' RETURNING ' + ', '.join(map(str, self.returning))
- return 'INSERT INTO %s' % self.table + columns + values + returning
+ with AliasManager():
+ return (self._with_str()
+ + 'INSERT INTO %s' % self.table + columns
+ + values + returning)
@property
def params(self):
p = []
- if isinstance(self.values, list):
- for values in self.values:
- for value in values:
- if isinstance(value, (Expression, Select)):
- p.extend(value.params)
- else:
- p.append(value)
- elif isinstance(self.values, Select):
+ p.extend(self._with_params())
+ if isinstance(self.values, Query):
p.extend(self.values.params)
if self.returning:
for exp in self.returning:
@@ -487,37 +564,37 @@ class Insert(Query):
class Update(Insert):
- __slots__ = ('__where', '__values', 'from_')
+ __slots__ = ('_where', '_values', 'from_')
def __init__(self, table, columns, values, from_=None, where=None,
- returning=None):
+ returning=None, **kwargs):
super(Update, self).__init__(table, columns=columns, values=values,
- returning=returning)
- self.__where = None
+ returning=returning, **kwargs)
+ self._where = None
self.from_ = From(from_) if from_ else None
self.where = where
@property
def values(self):
- return self.__values
+ return self._values
@values.setter
def values(self, value):
if isinstance(value, Select):
value = [value]
assert isinstance(value, list)
- self.__values = value
+ self._values = value
@property
def where(self):
- return self.__where
+ return self._where
@where.setter
def where(self, value):
from sql.operators import And, Or
if value is not None:
assert isinstance(value, (Expression, And, Or))
- self.__where = value
+ self._where = value
def __str__(self):
assert all(col.table == self.table for col in self.columns)
@@ -540,12 +617,14 @@ class Update(Insert):
returning = ''
if self.returning:
returning = ' RETURNING ' + ', '.join(map(str, self.returning))
- return ('UPDATE %s SET ' % table + values + from_ + where
- + returning)
+ return (self._with_str()
+ + 'UPDATE %s SET ' % table + values + from_
+ + where + returning)
@property
def params(self):
p = []
+ p.extend(self._with_params())
for value in self.values:
if isinstance(value, (Expression, Select)):
p.extend(value.params)
@@ -561,63 +640,68 @@ class Update(Insert):
return tuple(p)
-class Delete(Query):
- __slots__ = ('__table', '__where', '__returning', 'only')
+class Delete(WithQuery):
+ __slots__ = ('_table', '_where', '_returning', 'only')
def __init__(self, table, only=False, using=None, where=None,
- returning=None):
- self.__table = None
- self.__where = None
- self.__returning = None
+ returning=None, **kwargs):
+ self._table = None
+ self._where = None
+ self._returning = None
self.table = table
self.only = only
# TODO using (not standard)
self.where = where
self.returning = returning
+ super(Delete, self).__init__(**kwargs)
@property
def table(self):
- return self.__table
+ return self._table
@table.setter
def table(self, value):
assert isinstance(value, Table)
- self.__table = value
+ self._table = value
@property
def where(self):
- return self.__where
+ return self._where
@where.setter
def where(self, value):
from sql.operators import And, Or
if value is not None:
assert isinstance(value, (Expression, And, Or))
- self.__where = value
+ self._where = value
@property
def returning(self):
- return self.__returning
+ return self._returning
@returning.setter
def returning(self, value):
if value is not None:
assert isinstance(value, list)
- self.__returning = value
+ self._returning = value
def __str__(self):
- only = ' ONLY' if self.only else ''
- where = ''
- if self.where:
- where = ' WHERE ' + str(self.where)
- returning = ''
- if self.returning:
- returning = ' RETURNING ' + ', '.join(map(str, self.returning))
- return 'DELETE FROM%s %s' % (only, self.table) + where + returning
+ with AliasManager(exclude=[self.table]):
+ only = ' ONLY' if self.only else ''
+ where = ''
+ if self.where:
+ where = ' WHERE ' + str(self.where)
+ returning = ''
+ if self.returning:
+ returning = ' RETURNING ' + ', '.join(map(str, self.returning))
+ return (self._with_str()
+ + 'DELETE FROM%s %s' % (only, self.table)
+ + where + returning)
@property
def params(self):
p = []
+ p.extend(self._with_params())
if self.where:
p.extend(self.where.params)
if self.returning:
@@ -626,12 +710,12 @@ class Delete(Query):
return tuple(p)
-class CombiningQuery(Query, FromItem, _SelectQueryMixin):
+class CombiningQuery(FromItem, SelectQuery):
__slots__ = ('queries', 'all_')
_operator = ''
def __init__(self, *queries, **kwargs):
- assert all(isinstance(q, _SelectQueryMixin) for q in queries)
+ assert all(isinstance(q, Query) for q in queries)
self.queries = queries
self.all_ = kwargs.pop('all_', False)
super(CombiningQuery, self).__init__(**kwargs)
@@ -658,51 +742,68 @@ class Union(CombiningQuery):
_operator = 'UNION'
-class Interesect(CombiningQuery):
+class Intersect(CombiningQuery):
__slots__ = ()
_operator = 'INTERSECT'
+class Interesect(Intersect):
+ def __init__(self, *args, **kwargs):
+ warnings.warn('Interesect query is deprecated, use Intersect',
+ DeprecationWarning, stacklevel=2)
+ super(Interesect, self).__init__(*args, **kwargs)
+
+
class Except(CombiningQuery):
__slots__ = ()
_operator = 'EXCEPT'
class Table(FromItem):
- __slots__ = '__name'
+ __slots__ = ('_name', '_schema', '_database')
- def __init__(self, name):
+ def __init__(self, name, schema=None, database=None):
super(Table, self).__init__()
- self.__name = name
+ self._name = name
+ self._schema = schema
+ self._database = database
def __str__(self):
- return '"%s"' % self.__name
+ if self._database:
+ return '"%s"."%s"."%s"' % (
+ self._database, self._schema, self._name)
+ elif self._schema:
+ return '"%s"."%s"' % (self._schema, self._name)
+ else:
+ return '"%s"' % self._name
@property
def params(self):
return ()
- def insert(self, columns=None, values=None, returning=None):
+ def insert(self, columns=None, values=None, returning=None, with_=None):
return Insert(self, columns=columns, values=values,
- returning=returning)
+ returning=returning, with_=with_)
- def update(self, columns, values, from_=None, where=None, returning=None):
+ def update(self, columns, values, from_=None, where=None, returning=None,
+ with_=None):
return Update(self, columns=columns, values=values, from_=from_,
- where=where, returning=returning)
+ where=where, returning=returning, with_=with_)
- def delete(self, only=False, using=None, where=None, returning=None):
+ def delete(self, only=False, using=None, where=None, returning=None,
+ with_=None):
return Delete(self, only=only, using=using, where=where,
- returning=returning)
+ returning=returning, with_=with_)
class Join(FromItem):
- __slots__ = ('__left', '__right', '__condition', '__type_')
+ __slots__ = ('_left', '_right', '_condition', '_type_')
def __init__(self, left, right, type_='INNER', condition=None):
super(Join, self).__init__()
- self.__left, self.__right = None, None
- self.__condition = None
- self.__type_ = None
+ self._left, self._right = None, None
+ self._condition = None
+ self._type_ = None
self.left = left
self.right = right
self.condition = condition
@@ -710,43 +811,43 @@ class Join(FromItem):
@property
def left(self):
- return self.__left
+ return self._left
@left.setter
def left(self, value):
assert isinstance(value, FromItem)
- self.__left = value
+ self._left = value
@property
def right(self):
- return self.__right
+ return self._right
@right.setter
def right(self, value):
assert isinstance(value, FromItem)
- self.__right = value
+ self._right = value
@property
def condition(self):
- return self.__condition
+ return self._condition
@condition.setter
def condition(self, value):
from sql.operators import And, Or
if value is not None:
assert isinstance(value, (Expression, And, Or))
- self.__condition = value
+ self._condition = value
@property
def type_(self):
- return self.__type_
+ return self._type_
@type_.setter
def type_(self, value):
value = value.upper()
assert value in ('INNER', 'LEFT', 'LEFT OUTER',
'RIGHT', 'RIGHT OUTER', 'FULL', 'FULL OUTER', 'CROSS')
- self.__type_ = value
+ self._type_ = value
def __str__(self):
join = '%s %s JOIN %s' % (From([self.left]), self.type_,
@@ -786,21 +887,22 @@ class From(list):
def __str__(self):
def format(from_):
+ template = '%s'
if isinstance(from_, Query):
- return '(%s) AS "%s"' % (from_, from_.alias)
+ template = '(%s)'
+ alias = getattr(from_, 'alias', None)
+ # TODO column_alias
+ columns_definitions = getattr(from_, 'columns_definitions',
+ None)
+ # XXX find a better test for __getattr__ which returns Column
+ if (alias and columns_definitions
+ and not isinstance(columns_definitions, Column)):
+ return (template + ' AS "%s" (%s)') % (from_, alias,
+ columns_definitions)
+ elif alias:
+ return (template + ' AS "%s"') % (from_, alias)
else:
- alias = getattr(from_, 'alias', None)
- columns_definitions = getattr(from_, 'columns_definitions',
- None)
- # XXX find a better test for __getattr__ which returns Column
- if (alias and columns_definitions
- and not isinstance(columns_definitions, Column)):
- return '%s AS "%s" (%s)' % (from_, alias,
- columns_definitions)
- elif alias:
- return '%s AS "%s"' % (from_, alias)
- else:
- return str(from_)
+ return template % from_
return ', '.join(map(format, self))
@property
@@ -816,6 +918,35 @@ class From(list):
return From(super(From, self).__add__([other]))
+class Values(list, Query, FromItem):
+ __slots__ = ()
+
+ # TODO order, fetch
+
+ def __str__(self):
+ param = Flavor.get().param
+
+ def format_(value):
+ if isinstance(value, Expression):
+ return str(value)
+ else:
+ return param
+ return 'VALUES ' + ', '.join(
+ '(%s)' % ', '.join(map(format_, v))
+ for v in self)
+
+ @property
+ def params(self):
+ p = []
+ for values in self:
+ for value in values:
+ if isinstance(value, Expression):
+ p.extend(value.params)
+ else:
+ p.append(value)
+ return tuple(p)
+
+
class Expression(object):
__slots__ = ()
@@ -943,53 +1074,53 @@ class Expression(object):
class Literal(Expression):
- __slots__ = ('__value')
+ __slots__ = ('_value')
def __init__(self, value):
super(Literal, self).__init__()
- self.__value = value
+ self._value = value
@property
def value(self):
- return self.__value
+ return self._value
def __str__(self):
return Flavor.get().param
@property
def params(self):
- return (self.__value,)
+ return (self._value,)
Null = None
class Column(Expression):
- __slots__ = ('__from', '__name')
+ __slots__ = ('_from', '_name')
def __init__(self, from_, name):
super(Column, self).__init__()
- self.__from = from_
- self.__name = name
+ self._from = from_
+ self._name = name
@property
def table(self):
- return self.__from
+ return self._from
@property
def name(self):
- return self.__name
+ return self._name
def __str__(self):
- if self.__name == '*':
+ if self._name == '*':
t = '%s'
else:
t = '"%s"'
- alias = self.__from.alias
+ alias = self._from.alias
if alias:
t = '"%s".' + t
- return t % (alias, self.__name)
+ return t % (alias, self._name)
else:
- return t % self.__name
+ return t % self._name
@property
def params(self):
@@ -1063,35 +1194,35 @@ class Desc(Order):
class For(object):
- __slots__ = ('__tables', '__type_', 'nowait')
+ __slots__ = ('_tables', '_type_', 'nowait')
def __init__(self, type_, *tables, **kwargs):
- self.__tables = None
- self.__type_ = None
+ self._tables = None
+ self._type_ = None
self.tables = list(tables)
self.type_ = type_
self.nowait = kwargs.get('nowait')
@property
def tables(self):
- return self.__tables
+ return self._tables
@tables.setter
def tables(self, value):
if not isinstance(value, list):
value = [value]
all(isinstance(table, Table) for table in value)
- self.__tables = value
+ self._tables = value
@property
def type_(self):
- return self.__type_
+ return self._type_
@type_.setter
def type_(self, value):
value = value.upper()
assert value in ('UPDATE', 'SHARE')
- self.__type_ = value
+ self._type_ = value
def __str__(self):
tables = ''
diff --git a/sql/functions.py b/sql/functions.py
index c987df2..1eeb7c8 100644
--- a/sql/functions.py
+++ b/sql/functions.py
@@ -47,7 +47,7 @@ __all__ = ['Abs', 'Cbrt', 'Ceil', 'Degrees', 'Div', 'Exp', 'Floor', 'Ln',
class Function(Expression, FromItem):
- __slots__ = ('args', '__columns_definitions')
+ __slots__ = ('args', '_columns_definitions')
table = ''
name = ''
_function = ''
@@ -59,12 +59,12 @@ class Function(Expression, FromItem):
@property
def columns_definitions(self):
return ', '.join('"%s" %s' % (c, d)
- for c, d in self.__columns_definitions)
+ for c, d in self._columns_definitions)
@columns_definitions.setter
def columns_definitions(self, value):
assert isinstance(value, list)
- self.__columns_definitions = value
+ self._columns_definitions = value
@staticmethod
def _format(value):
diff --git a/sql/operators.py b/sql/operators.py
index 25d9157..3072b90 100644
--- a/sql/operators.py
+++ b/sql/operators.py
@@ -257,7 +257,14 @@ class FloorDiv(BinaryOperator):
class Mod(BinaryOperator):
__slots__ = ()
- _operator = '%'
+
+ @property
+ def _operator(self):
+ # '%' must be escaped with format paramstyle
+ if Flavor.get().paramstyle == 'format':
+ return '%%'
+ else:
+ return '%'
class Pow(BinaryOperator):
diff --git a/sql/tests/test_delete.py b/sql/tests/test_combining_query.py
similarity index 53%
copy from sql/tests/test_delete.py
copy to sql/tests/test_combining_query.py
index 4254f44..038d522 100644
--- a/sql/tests/test_delete.py
+++ b/sql/tests/test_combining_query.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
-# Copyright (c) 2011-2013, Cédric Krier
-# Copyright (c) 2011-2013, B2CK
+# Copyright (c) 2014, Cédric Krier
+# Copyright (c) 2014, B2CK
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
@@ -28,32 +28,46 @@
import unittest
-from sql import Table
+from sql import Table, Union
-class TestDelete(unittest.TestCase):
- table = Table('t')
+class TestUnion(unittest.TestCase):
+ q1 = Table('t1').select()
+ q2 = Table('t2').select()
+ q3 = Table('t3').select()
- def test_delete1(self):
- query = self.table.delete()
- self.assertEqual(str(query), 'DELETE FROM "t"')
+ def test_union2(self):
+ query = Union(self.q1, self.q2)
+ self.assertEqual(str(query),
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b"')
+ self.assertEqual(query.params, ())
+
+ query = self.q1 | self.q2
+ self.assertEqual(str(query),
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b"')
+ self.assertEqual(query.params, ())
+
+ def test_union3(self):
+ query = Union(self.q1, self.q2, self.q3)
+ self.assertEqual(str(query),
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b" '
+ 'UNION SELECT * FROM "t3" AS "c"')
self.assertEqual(query.params, ())
- def test_delete2(self):
- query = self.table.delete(where=(self.table.c == 'foo'))
- self.assertEqual(str(query), 'DELETE FROM "t" WHERE ("c" = %s)')
- self.assertEqual(query.params, ('foo',))
+ query = Union(Union(self.q1, self.q2), self.q3)
+ self.assertEqual(str(query),
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b" '
+ 'UNION SELECT * FROM "t3" AS "c"')
+ self.assertEqual(query.params, ())
- def test_delete3(self):
- t1 = Table('t1')
- t2 = Table('t2')
- query = t1.delete(where=(t1.c.in_(t2.select(t2.c))))
+ query = Union(self.q1, Union(self.q2, self.q3))
self.assertEqual(str(query),
- 'DELETE FROM "t1" WHERE ("c" IN ('
- 'SELECT "a"."c" FROM "t2" AS "a"))')
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b" '
+ 'UNION SELECT * FROM "t3" AS "c"')
self.assertEqual(query.params, ())
- def test_delete_returning(self):
- query = self.table.delete(returning=[self.table.c])
- self.assertEqual(str(query), 'DELETE FROM "t" RETURNING "c"')
+ query = self.q1 | self.q2 | self.q3
+ self.assertEqual(str(query),
+ 'SELECT * FROM "t1" AS "a" UNION SELECT * FROM "t2" AS "b" '
+ 'UNION SELECT * FROM "t3" AS "c"')
self.assertEqual(query.params, ())
diff --git a/sql/tests/test_delete.py b/sql/tests/test_delete.py
index 4254f44..e629cf0 100644
--- a/sql/tests/test_delete.py
+++ b/sql/tests/test_delete.py
@@ -28,7 +28,7 @@
import unittest
-from sql import Table
+from sql import Table, With
class TestDelete(unittest.TestCase):
@@ -57,3 +57,14 @@ class TestDelete(unittest.TestCase):
query = self.table.delete(returning=[self.table.c])
self.assertEqual(str(query), 'DELETE FROM "t" RETURNING "c"')
self.assertEqual(query.params, ())
+
+ def test_with(self):
+ t1 = Table('t1')
+ w = With(query=t1.select(t1.c1))
+
+ query = self.table.delete(with_=[w],
+ where=self.table.c2.in_(w.select(w.c3)))
+ self.assertEqual(str(query),
+ 'WITH a AS (SELECT "b"."c1" FROM "t1" AS "b") '
+ 'DELETE FROM "t" WHERE ("c2" IN (SELECT "a"."c3" FROM a AS "a"))')
+ self.assertEqual(query.params, ())
diff --git a/sql/tests/test_insert.py b/sql/tests/test_insert.py
index 283fa80..a869de3 100644
--- a/sql/tests/test_insert.py
+++ b/sql/tests/test_insert.py
@@ -29,7 +29,7 @@
import unittest
-from sql import Table
+from sql import Table, With
from sql.functions import Abs
@@ -78,3 +78,16 @@ class TestInsert(unittest.TestCase):
'INSERT INTO "t" ("c1", "c2") VALUES (%s, %s) '
'RETURNING "c1", "c2"')
self.assertEqual(query.params, ('foo', 'bar'))
+
+ def test_with(self):
+ t1 = Table('t1')
+ w = With(query=t1.select())
+
+ query = self.table.insert(
+ [self.table.c1],
+ with_=[w],
+ values=w.select())
+ self.assertEqual(str(query),
+ 'WITH a AS (SELECT * FROM "t1" AS "b") '
+ 'INSERT INTO "t" ("c1") SELECT * FROM a AS "a"')
+ self.assertEqual(query.params, ())
diff --git a/sql/tests/test_operators.py b/sql/tests/test_operators.py
index 23619e3..84fc218 100644
--- a/sql/tests/test_operators.py
+++ b/sql/tests/test_operators.py
@@ -193,8 +193,27 @@ class TestOperators(unittest.TestCase):
def test_mod(self):
for mod in [Mod(self.table.c1, self.table.c2),
self.table.c1 % self.table.c2]:
+ self.assertEqual(str(mod), '("c1" %% "c2")')
+ self.assertEqual(mod.params, ())
+
+ def test_mod_paramstyle(self):
+ flavor = Flavor(paramstyle='format')
+ Flavor.set(flavor)
+ try:
+ mod = Mod(self.table.c1, self.table.c2)
+ self.assertEqual(str(mod), '("c1" %% "c2")')
+ self.assertEqual(mod.params, ())
+ finally:
+ Flavor.set(Flavor())
+
+ flavor = Flavor(paramstyle='qmark')
+ Flavor.set(flavor)
+ try:
+ mod = Mod(self.table.c1, self.table.c2)
self.assertEqual(str(mod), '("c1" % "c2")')
self.assertEqual(mod.params, ())
+ finally:
+ Flavor.set(Flavor())
def test_pow(self):
for pow_ in [Pow(self.table.c1, self.table.c2),
diff --git a/sql/tests/test_select.py b/sql/tests/test_select.py
index bf8cd26..6fbea26 100644
--- a/sql/tests/test_select.py
+++ b/sql/tests/test_select.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2011-2013, Cédric Krier
-# Copyright (c) 2013, Nicolas Évrard
+# Copyright (c) 2013-2014, Nicolas Évrard
# Copyright (c) 2011-2013, B2CK
# All rights reserved.
#
@@ -28,8 +28,10 @@
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import unittest
+import warnings
+from copy import deepcopy
-from sql import Table, Join, Union, Literal, Flavor, For
+from sql import Table, Join, Union, Literal, Flavor, For, With
from sql.functions import Now, Function
from sql.aggregate import Min
@@ -107,6 +109,16 @@ class TestSelect(unittest.TestCase):
self.assertEqual(str(intersect),
'SELECT * FROM "t" AS "a" INTERSECT SELECT * FROM "t2" AS "b"')
+ from sql import Intersect, Interesect
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter('always')
+ interesect = Interesect(query1, query2)
+ self.assertEqual(len(w), 1)
+ self.assertTrue(issubclass(w[-1].category, DeprecationWarning))
+ self.assertIn('Interesect query is deprecated, use Intersect',
+ str(w[-1].message))
+ self.assertTrue(isinstance(interesect, Intersect))
+
def test_select_except(self):
query1 = self.table.select()
query2 = Table('t2').select()
@@ -224,3 +236,19 @@ class TestSelect(unittest.TestCase):
self.assertEqual(str(query),
'SELECT "a"."c" FROM "t" AS "a" FOR UPDATE')
self.assertEqual(query.params, ())
+
+ def test_copy(self):
+ query = self.table.select()
+ copy_query = deepcopy(query)
+ self.assertNotEqual(query, copy_query)
+ self.assertEqual(str(copy_query), 'SELECT * FROM "t" AS "a"')
+ self.assertEqual(copy_query.params, ())
+
+ def test_with(self):
+ w = With(query=self.table.select(self.table.c1))
+
+ query = w.select(with_=[w])
+ self.assertEqual(str(query),
+ 'WITH a AS (SELECT "b"."c1" FROM "t" AS "b") '
+ 'SELECT * FROM a AS "a"')
+ self.assertEqual(query.params, ())
diff --git a/sql/tests/test_delete.py b/sql/tests/test_table.py
similarity index 60%
copy from sql/tests/test_delete.py
copy to sql/tests/test_table.py
index 4254f44..2d26163 100644
--- a/sql/tests/test_delete.py
+++ b/sql/tests/test_table.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
-# Copyright (c) 2011-2013, Cédric Krier
-# Copyright (c) 2011-2013, B2CK
+# Copyright (c) 2015, Cédric Krier
+# Copyright (c) 2015, B2CK
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
@@ -26,34 +26,22 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
import unittest
from sql import Table
-class TestDelete(unittest.TestCase):
- table = Table('t')
-
- def test_delete1(self):
- query = self.table.delete()
- self.assertEqual(str(query), 'DELETE FROM "t"')
- self.assertEqual(query.params, ())
-
- def test_delete2(self):
- query = self.table.delete(where=(self.table.c == 'foo'))
- self.assertEqual(str(query), 'DELETE FROM "t" WHERE ("c" = %s)')
- self.assertEqual(query.params, ('foo',))
-
- def test_delete3(self):
- t1 = Table('t1')
- t2 = Table('t2')
- query = t1.delete(where=(t1.c.in_(t2.select(t2.c))))
- self.assertEqual(str(query),
- 'DELETE FROM "t1" WHERE ("c" IN ('
- 'SELECT "a"."c" FROM "t2" AS "a"))')
- self.assertEqual(query.params, ())
-
- def test_delete_returning(self):
- query = self.table.delete(returning=[self.table.c])
- self.assertEqual(str(query), 'DELETE FROM "t" RETURNING "c"')
- self.assertEqual(query.params, ())
+class TestTable(unittest.TestCase):
+
+ def test_name(self):
+ t = Table('mytable')
+ self.assertEqual(str(t), '"mytable"')
+
+ def test_schema(self):
+ t = Table('mytable', schema='myschema')
+ self.assertEqual(str(t), '"myschema"."mytable"')
+
+ def test_database(self):
+ t = Table('mytable', database='mydatabase', schema='myschema')
+ self.assertEqual(str(t), '"mydatabase"."myschema"."mytable"')
diff --git a/sql/tests/test_update.py b/sql/tests/test_update.py
index 63ef2a8..a1cbb3c 100644
--- a/sql/tests/test_update.py
+++ b/sql/tests/test_update.py
@@ -28,7 +28,7 @@
import unittest
-from sql import Table, Literal
+from sql import Table, Literal, With
class TestUpdate(unittest.TestCase):
@@ -70,3 +70,17 @@ class TestUpdate(unittest.TestCase):
self.assertEqual(str(query),
'UPDATE "t" SET "c" = %s RETURNING "t"."c"')
self.assertEqual(query.params, ('foo',))
+
+ def test_with(self):
+ t1 = Table('t1')
+ w = With(query=t1.select(t1.c1))
+
+ query = self.table.update(
+ [self.table.c2],
+ with_=[w],
+ values=[w.select(w.c3, where=w.c4 == 2)])
+ self.assertEqual(str(query),
+ 'WITH b AS (SELECT "c"."c1" FROM "t1" AS "c") '
+ 'UPDATE "t" SET "c2" = (SELECT "b"."c3" FROM b AS "b" '
+ 'WHERE ("b"."c4" = %s))')
+ self.assertEqual(query.params, (2,))
diff --git a/sql/tests/test_delete.py b/sql/tests/test_values.py
similarity index 60%
copy from sql/tests/test_delete.py
copy to sql/tests/test_values.py
index 4254f44..8885f93 100644
--- a/sql/tests/test_delete.py
+++ b/sql/tests/test_values.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
#
-# Copyright (c) 2011-2013, Cédric Krier
-# Copyright (c) 2011-2013, B2CK
+# Copyright (c) 2014, Cédric Krier
+# Copyright (c) 2014, B2CK
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
@@ -28,32 +28,29 @@
import unittest
-from sql import Table
+from sql import Values
-class TestDelete(unittest.TestCase):
- table = Table('t')
+class TestValues(unittest.TestCase):
+ def test_single_values(self):
+ values = Values([[1]])
+ self.assertEqual(str(values), 'VALUES (%s)')
+ self.assertEqual(values.params, (1,))
- def test_delete1(self):
- query = self.table.delete()
- self.assertEqual(str(query), 'DELETE FROM "t"')
- self.assertEqual(query.params, ())
+ def test_many_values(self):
+ values = Values([[1, 2], [3, 4]])
+ self.assertEqual(str(values), 'VALUES (%s, %s), (%s, %s)')
+ self.assertEqual(values.params, (1, 2, 3, 4))
- def test_delete2(self):
- query = self.table.delete(where=(self.table.c == 'foo'))
- self.assertEqual(str(query), 'DELETE FROM "t" WHERE ("c" = %s)')
- self.assertEqual(query.params, ('foo',))
-
- def test_delete3(self):
- t1 = Table('t1')
- t2 = Table('t2')
- query = t1.delete(where=(t1.c.in_(t2.select(t2.c))))
+ def test_select(self):
+ values = Values([[1], [2], [3]])
+ query = values.select()
self.assertEqual(str(query),
- 'DELETE FROM "t1" WHERE ("c" IN ('
- 'SELECT "a"."c" FROM "t2" AS "a"))')
- self.assertEqual(query.params, ())
-
- def test_delete_returning(self):
- query = self.table.delete(returning=[self.table.c])
- self.assertEqual(str(query), 'DELETE FROM "t" RETURNING "c"')
- self.assertEqual(query.params, ())
+ 'SELECT * FROM (VALUES (%s), (%s), (%s)) AS "a"')
+ self.assertEqual(query.params, (1, 2, 3))
+
+ def test_union(self):
+ values = Values([[1]])
+ values |= Values([[2]])
+ self.assertEqual(str(values), 'VALUES (%s) UNION VALUES (%s)')
+ self.assertEqual(values.params, (1, 2))
diff --git a/sql/tests/test_with.py b/sql/tests/test_with.py
new file mode 100644
index 0000000..487c722
--- /dev/null
+++ b/sql/tests/test_with.py
@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2014, Nicolas Évrard
+# Copyright (c) 2014, B2CK
+# 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.
+# * Neither the name of the <organization> nor the
+# names of its contributors may 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 <COPYRIGHT HOLDER> 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.
+
+import unittest
+
+from sql import AliasManager, Table, Literal, Values, With, WithQuery
+
+
+class TestWith(unittest.TestCase):
+ table = Table('t')
+
+ def test_with(self):
+ with AliasManager():
+ simple = With(query=self.table.select(self.table.id,
+ where=self.table.id == 1))
+
+ self.assertEqual(simple.statement(),
+ 'a AS ('
+ 'SELECT "b"."id" FROM "t" AS "b" WHERE ("b"."id" = %s)'
+ ')')
+ self.assertEqual(simple.statement_params(), (1,))
+
+ def test_with_columns(self):
+ with AliasManager():
+ second = With('a', query=self.table.select(self.table.a))
+
+ self.assertEqual(second.statement(),
+ 'a("a") AS ('
+ 'SELECT "b"."a" FROM "t" AS "b"'
+ ')')
+ self.assertEqual(second.statement_params(), ())
+
+ def test_with_query(self):
+ with AliasManager():
+ simple = With()
+ simple.query = self.table.select(self.table.id,
+ where=self.table.id == 1)
+ second = With()
+ second.query = simple.select()
+
+ wq = WithQuery(with_=[simple, second])
+ self.assertEqual(wq._with_str(),
+ 'WITH a AS ('
+ 'SELECT "b"."id" FROM "t" AS "b" WHERE ("b"."id" = %s)'
+ '), c AS ('
+ 'SELECT * FROM a AS "a"'
+ ') ')
+ self.assertEqual(wq._with_params(), (1,))
+
+ def test_recursive(self):
+ upto10 = With('n', recursive=True)
+ upto10.query = Values([(1,)])
+ upto10.query |= upto10.select(
+ upto10.n + Literal(1),
+ where=upto10.n < Literal(100))
+ upto10.query.all_ = True
+
+ q = upto10.select(with_=[upto10])
+ self.assertEqual(str(q),
+ 'WITH RECURSIVE a("n") AS ('
+ 'VALUES (%s) '
+ 'UNION ALL '
+ 'SELECT ("a"."n" + %s) FROM a AS "a" WHERE ("a"."n" < %s)'
+ ') SELECT * FROM a AS "a"')
+ self.assertEqual(q.params, (1, 1, 100))
--
python-sql
More information about the tryton-debian-vcs
mailing list