=======
CHANGES
=======
+
0.5.4
=====
+
- orm
- - Fixed the "set collection" function on "dynamic" relations
- to initiate events correctly. Previously a collection
- could only be assigned to a pending parent instance,
- otherwise modified events would not be fired correctly.
- Set collection is now compatible with merge(),
- fixes [ticket:1352].
-
- - Lazy loader will not use get() if the "lazy load"
- SQL clause matches the clause used by get(), but
- contains some parameters hardcoded. Previously
- the lazy strategy would fail with the get(). Ideally
- get() would be used with the hardcoded parameters
- but this would require further development.
+ - Fixed the "set collection" function on "dynamic" relations to
+ initiate events correctly. Previously a collection could only
+ be assigned to a pending parent instance, otherwise modified
+ events would not be fired correctly. Set collection is now
+ compatible with merge(), fixes [ticket:1352].
+
+ - Lazy loader will not use get() if the "lazy load" SQL clause
+ matches the clause used by get(), but contains some parameters
+ hardcoded. Previously the lazy strategy would fail with the
+ get(). Ideally get() would be used with the hardcoded
+ parameters but this would require further development.
[ticket:1357]
- sql
- - Fixed __repr__() and other _get_colspec() methods on
+ - ``sqlalchemy.extract()`` is now dialect sensitive and can
+ extract components of timestamps idiomatically across the
+ supported databases, including SQLite.
+
+ - Fixed __repr__() and other _get_colspec() methods on
ForeignKey constructed from __clause_element__() style
construct (i.e. declarative columns). [ticket:1353]
-
+
- mssql
- Corrected problem with information schema not working with a
- binary collation based database. Cleaned up information
- schema since it is only used by mssql now. [ticket:1343]
+ binary collation based database. Cleaned up information schema
+ since it is only used by mssql now. [ticket:1343]
0.5.3
=====
class AccessCompiler(compiler.DefaultCompiler):
+ extract_map = compiler.DefaultCompiler.extract_map.copy()
+ extract_map.update ({
+ 'month': 'm',
+ 'day': 'd',
+ 'year': 'yyyy',
+ 'second': 's',
+ 'hour': 'h',
+ 'doy': 'y',
+ 'minute': 'n',
+ 'quarter': 'q',
+ 'dow': 'w',
+ 'week': 'ww'
+ })
+
def visit_select_precolumns(self, select):
"""Access puts TOP, it's version of LIMIT here """
s = select.distinct and "DISTINCT " or ""
return (self.process(join.left, asfrom=True) + (join.isouter and " LEFT OUTER JOIN " or " INNER JOIN ") + \
self.process(join.right, asfrom=True) + " ON " + self.process(join.onclause))
+ def visit_extract(self, extract):
+ field = self.extract_map.get(extract.field, extract.field)
+ return 'DATEPART("%s", %s)' % (field, self.process(extract.expr))
+
class AccessSchemaGenerator(compiler.SchemaGenerator):
def get_column_specification(self, column, **kwargs):
dialect.schemadropper = AccessSchemaDropper
dialect.preparer = AccessIdentifierPreparer
dialect.defaultrunner = AccessDefaultRunner
-dialect.execution_ctx_cls = AccessExecutionContext
\ No newline at end of file
+dialect.execution_ctx_cls = AccessExecutionContext
}
)
+ extract_map = compiler.DefaultCompiler.extract_map.copy()
+ extract_map.update ({
+ 'doy': 'dayofyear',
+ 'dow': 'weekday',
+ 'milliseconds': 'millisecond',
+ 'microseconds': 'microsecond'
+ })
+
def __init__(self, *args, **kwargs):
super(MSSQLCompiler, self).__init__(*args, **kwargs)
self.tablealiases = {}
kwargs['mssql_aliased'] = True
return super(MSSQLCompiler, self).visit_alias(alias, **kwargs)
+ def visit_extract(self, extract):
+ field = self.extract_map.get(extract.field, extract.field)
+ return 'DATEPART("%s", %s)' % (field, self.process(extract.expr))
+
def visit_savepoint(self, savepoint_stmt):
util.warn("Savepoint support in mssql is experimental and may lead to data loss.")
return "SAVE TRANSACTION %s" % self.preparer.format_savepoint(savepoint_stmt)
"utc_timestamp":"UTC_TIMESTAMP"
})
+ extract_map = compiler.DefaultCompiler.extract_map.copy()
+ extract_map.update ({
+ 'milliseconds': 'millisecond',
+ })
def visit_typeclause(self, typeclause):
type_ = typeclause.type.dialect_impl(self.dialect)
else:
return text
+ def visit_extract(self, extract, **kwargs):
+ field = self.extract_map.get(extract.field, extract.field)
+ return "EXTRACT(%s FROM %s::timestamp)" % (
+ field, self.process(extract.expr))
+
+
class PGSchemaGenerator(compiler.SchemaGenerator):
def get_column_specification(self, column, **kwargs):
colspec = self.preparer.format_column(column)
}
)
+ extract_map = compiler.DefaultCompiler.extract_map.copy()
+ extract_map.update({
+ 'month': '%m',
+ 'day': '%d',
+ 'year': '%Y',
+ 'second': '%S',
+ 'hour': '%H',
+ 'doy': '%j',
+ 'minute': '%M',
+ 'epoch': '%s',
+ 'dow': '%w',
+ 'week': '%W'
+ })
+
def visit_cast(self, cast, **kwargs):
if self.dialect.supports_cast:
return super(SQLiteCompiler, self).visit_cast(cast)
else:
return self.process(cast.clause)
+ def visit_extract(self, extract):
+ try:
+ return "CAST(STRFTIME('%s', %s) AS INTEGER)" % (
+ self.extract_map[extract.field], self.process(extract.expr))
+ except KeyError:
+ raise exc.ArgumentError(
+ "%s is not a valid extract argument." % extract.field)
+
def limit_clause(self, select):
text = ""
if select._limit is not None:
sql_operators.mod: lambda x, y: "MOD(%s, %s)" % (x, y),
})
+ extract_map = compiler.DefaultCompiler.extract_map.copy()
+ extract_map.update ({
+ 'doy': 'dayofyear',
+ 'dow': 'weekday',
+ 'milliseconds': 'millisecond'
+ })
+
+
def bindparam_string(self, name):
res = super(SybaseSQLCompiler, self).bindparam_string(name)
if name.lower().startswith('literal'):
res = "CAST(%s AS %s)" % (res, self.process(cast.typeclause))
return res
+ def visit_extract(self, extract):
+ field = self.extract_map.get(extract.field, extract.field)
+ return 'DATEPART("%s", %s)' % (field, self.process(extract.expr))
+
def for_update_clause(self, select):
# "FOR UPDATE" is only allowed on "DECLARE CURSOR" which SQLAlchemy doesn't use
return ''
functions.user: 'USER'
}
+EXTRACT_MAP = {
+ 'month': 'month',
+ 'day': 'day',
+ 'year': 'year',
+ 'second': 'second',
+ 'hour': 'hour',
+ 'doy': 'doy',
+ 'minute': 'minute',
+ 'quarter': 'quarter',
+ 'dow': 'dow',
+ 'week': 'week',
+ 'epoch': 'epoch',
+ 'milliseconds': 'milliseconds',
+ 'microseconds': 'microseconds',
+ 'timezone_hour': 'timezone_hour',
+ 'timezone_minute': 'timezone_minute'
+}
class _CompileLabel(visitors.Visitable):
"""lightweight label object which acts as an expression._Label."""
operators = OPERATORS
functions = FUNCTIONS
+ extract_map = EXTRACT_MAP
# if we are insert/update/delete.
# set to true when we visit an INSERT, UPDATE or DELETE
def visit_cast(self, cast, **kwargs):
return "CAST(%s AS %s)" % (self.process(cast.clause), self.process(cast.typeclause))
+ def visit_extract(self, extract, **kwargs):
+ field = self.extract_map.get(extract.field, extract.field)
+ return "EXTRACT(%s FROM %s)" % (field, self.process(extract.expr))
+
def visit_function(self, func, result_map=None, **kwargs):
if result_map is not None:
result_map[func.name.lower()] = (func.name, None, func.type)
def extract(field, expr):
"""Return the clause ``extract(field FROM expr)``."""
- expr = _BinaryExpression(text(field), expr, operators.from_)
- return func.extract(expr)
+ return _Extract(field, expr)
def collate(expression, collation):
"""Return the clause ``expression COLLATE collation``."""
return self.clause._from_objects
+class _Extract(ColumnElement):
+
+ __visit_name__ = 'extract'
+
+ def __init__(self, field, expr, **kwargs):
+ self.type = sqltypes.Integer()
+ self.field = field
+ self.expr = _literal_as_binds(expr, None)
+
+ def _copy_internals(self, clone=_clone):
+ self.field = clone(self.field)
+ self.expr = clone(self.expr)
+
+ def get_children(self, **kwargs):
+ return self.field, self.expr
+
+ @property
+ def _from_objects(self):
+ return self.expr._from_objects
+
+
class _UnaryExpression(ColumnElement):
__visit_name__ = 'unary'
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
+from sqlalchemy import sql
from sqlalchemy.databases import access
from testlib import *
-class BasicTest(TestBase, AssertsExecutionResults):
- # A simple import of the database/ module should work on all systems.
- def test_import(self):
- # we got this far, right?
- return True
+class CompileTest(TestBase, AssertsCompiledSQL):
+ __dialect__ = access.dialect()
+
+ def test_extract(self):
+ t = sql.table('t', sql.column('col1'))
+
+ mapping = {
+ 'month': 'm',
+ 'day': 'd',
+ 'year': 'yyyy',
+ 'second': 's',
+ 'hour': 'h',
+ 'doy': 'y',
+ 'minute': 'n',
+ 'quarter': 'q',
+ 'dow': 'w',
+ 'week': 'ww'
+ }
+
+ for field, subst in mapping.items():
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ 'SELECT DATEPART("%s", t.col1) AS anon_1 FROM t' % subst)
if __name__ == "__main__":
self.assert_compile(func.current_date(), "GETDATE()")
self.assert_compile(func.length(3), "LEN(:length_1)")
+ def test_extract(self):
+ t = table('t', column('col1'))
+
+ for field in 'day', 'month', 'year':
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ 'SELECT DATEPART("%s", t.col1) AS anon_1 FROM t' % field)
+
class IdentityInsertTest(TestBase, AssertsCompiledSQL):
__only_on__ = 'mssql'
for type_, expected in specs:
self.assert_compile(cast(t.c.col, type_), expected)
+ def test_extract(self):
+ t = sql.table('t', sql.column('col1'))
+
+ for field in 'year', 'month', 'day':
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ "SELECT EXTRACT(%s FROM t.col1) AS anon_1 FROM t" % field)
+
+ # millsecondS to millisecond
+ self.assert_compile(
+ select([extract('milliseconds', t.c.col1)]),
+ "SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t")
+
class RawReflectionTest(TestBase):
def setUp(self):
assert dialect.identifier_preparer.format_sequence(seq) == '"Some_Schema"."My_Seq"'
class CompileTest(TestBase, AssertsCompiledSQL):
+ __dialect__ = postgres.dialect()
+
def test_update_returning(self):
dialect = postgres.dialect()
table1 = table('mytable',
i = insert(table1, values=dict(name='foo'), postgres_returning=[func.length(table1.c.name)])
self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) RETURNING length(mytable.name)", dialect=dialect)
+ def test_extract(self):
+ t = table('t', column('col1'))
+
+ for field in 'year', 'month', 'day':
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ "SELECT EXTRACT(%s FROM t.col1::timestamp) AS anon_1 "
+ "FROM t" % field)
+
+
class ReturningTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgres'
import testenv; testenv.configure_for_tests()
import datetime
from sqlalchemy import *
-from sqlalchemy import exc
+from sqlalchemy import exc, sql
from sqlalchemy.databases import sqlite
from testlib import *
pass
raise
+
+class SQLTest(TestBase, AssertsCompiledSQL):
+ """Tests SQLite-dialect specific compilation."""
+
+ __dialect__ = sqlite.dialect()
+
+
+ def test_extract(self):
+ t = sql.table('t', sql.column('col1'))
+
+ mapping = {
+ 'month': '%m',
+ 'day': '%d',
+ 'year': '%Y',
+ 'second': '%S',
+ 'hour': '%H',
+ 'doy': '%j',
+ 'minute': '%M',
+ 'epoch': '%s',
+ 'dow': '%w',
+ 'week': '%W',
+ }
+
+ for field, subst in mapping.items():
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ "SELECT CAST(STRFTIME('%s', t.col1) AS INTEGER) AS anon_1 "
+ "FROM t" % subst)
+
+
class InsertTest(TestBase, AssertsExecutionResults):
"""Tests inserts and autoincrement."""
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
+from sqlalchemy import sql
from sqlalchemy.databases import sybase
from testlib import *
-class BasicTest(TestBase, AssertsExecutionResults):
- # A simple import of the database/ module should work on all systems.
- def test_import(self):
- # we got this far, right?
- return True
+class CompileTest(TestBase, AssertsCompiledSQL):
+ __dialect__ = sybase.dialect()
+
+ def test_extract(self):
+ t = sql.table('t', sql.column('col1'))
+
+ mapping = {
+ 'day': 'day',
+ 'doy': 'dayofyear',
+ 'dow': 'weekday',
+ 'milliseconds': 'millisecond',
+ 'millisecond': 'millisecond',
+ 'year': 'year',
+ }
+
+ for field, subst in mapping.items():
+ self.assert_compile(
+ select([extract(field, t.c.col1)]),
+ 'SELECT DATEPART("%s", t.col1) AS anon_1 FROM t' % subst)
+
if __name__ == "__main__":
assert x == y == z == w == q == r
+ def test_extract_bind(self):
+ """Basic common denominator execution tests for extract()"""
+
+ date = datetime.date(2010, 5, 1)
+
+ def execute(field):
+ return testing.db.execute(select([extract(field, date)])).scalar()
+
+ assert execute('year') == 2010
+ assert execute('month') == 5
+ assert execute('day') == 1
+
+ date = datetime.datetime(2010, 5, 1, 12, 11, 10)
+
+ assert execute('year') == 2010
+ assert execute('month') == 5
+ assert execute('day') == 1
+
+ def test_extract_expression(self):
+ meta = MetaData(testing.db)
+ table = Table('test', meta,
+ Column('dt', DateTime),
+ Column('d', Date))
+ meta.create_all()
+ try:
+ table.insert().execute(
+ {'dt': datetime.datetime(2010, 5, 1, 12, 11, 10),
+ 'd': datetime.date(2010, 5, 1) })
+ rs = select([extract('year', table.c.dt),
+ extract('month', table.c.d)]).execute()
+ row = rs.fetchone()
+ assert row[0] == 2010
+ assert row[1] == 5
+ rs.close()
+ finally:
+ meta.drop_all()
+
+
def exec_sorted(statement, *args, **kw):
"""Executes a statement and returns a sorted list plain tuple rows."""
"SELECT values.id FROM values WHERE values.val1 / (values.val2 - values.val1) / values.val1 > :param_1"
)
- def test_extract(self):
- """test the EXTRACT function"""
- self.assert_compile(select([extract("month", table3.c.otherstuff)]), "SELECT extract(month FROM thirdtable.otherstuff) AS extract_1 FROM thirdtable")
-
- self.assert_compile(select([extract("day", func.to_date("03/20/2005", "MM/DD/YYYY"))]), "SELECT extract(day FROM to_date(:to_date_1, :to_date_2)) AS extract_1")
-
def test_collate(self):
for expr in (select([table1.c.name.collate('latin1_german2_ci')]),
select([collate(table1.c.name, 'latin1_german2_ci')])):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
