# -*- encoding: utf-8
from sqlalchemy.testing import eq_
-from sqlalchemy import *
from sqlalchemy import schema
from sqlalchemy.sql import table, column
from sqlalchemy.databases import mssql
from sqlalchemy.dialects.mssql import mxodbc
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from sqlalchemy import sql
+from sqlalchemy import Integer, String, Table, Column, select, MetaData,\
+ update, delete, insert, extract, union, func, PrimaryKeyConstraint, \
+ UniqueConstraint, Index, Sequence, literal
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
def test_select_with_nolock(self):
t = table('sometable', column('somecolumn'))
- self.assert_compile(t.select().with_hint(t, 'WITH (NOLOCK)'),
- 'SELECT sometable.somecolumn FROM sometable WITH (NOLOCK)')
+ self.assert_compile(
+ t.select().with_hint(t, 'WITH (NOLOCK)'),
+ 'SELECT sometable.somecolumn FROM sometable WITH (NOLOCK)')
def test_join_with_hint(self):
t1 = table('t1',
- column('a', Integer),
- column('b', String),
- column('c', String),
- )
+ column('a', Integer),
+ column('b', String),
+ column('c', String),
+ )
t2 = table('t2',
- column("a", Integer),
- column("b", Integer),
- column("c", Integer),
- )
- join = t1.join(t2, t1.c.a==t2.c.a).\
- select().with_hint(t1, 'WITH (NOLOCK)')
+ column("a", Integer),
+ column("b", Integer),
+ column("c", Integer),
+ )
+ join = t1.join(t2, t1.c.a == t2.c.a).\
+ select().with_hint(t1, 'WITH (NOLOCK)')
self.assert_compile(
join,
'SELECT t1.a, t1.b, t1.c, t2.a, t2.b, t2.c '
for darg in ("*", "mssql"):
self.assert_compile(
t.insert().
- values(somecolumn="x").
- with_hint("WITH (PAGLOCK)",
- selectable=targ,
- dialect_name=darg),
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
"INSERT INTO sometable WITH (PAGLOCK) "
"(somecolumn) VALUES (:somecolumn)"
)
for targ in (None, t):
for darg in ("*", "mssql"):
self.assert_compile(
- t.update().where(t.c.somecolumn=="q").
- values(somecolumn="x").
- with_hint("WITH (PAGLOCK)",
- selectable=targ,
- dialect_name=darg),
+ t.update().where(t.c.somecolumn == "q").
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
"UPDATE sometable WITH (PAGLOCK) "
"SET somecolumn=:somecolumn "
"WHERE sometable.somecolumn = :somecolumn_1"
def test_update_exclude_hint(self):
t = table('sometable', column('somecolumn'))
self.assert_compile(
- t.update().where(t.c.somecolumn=="q").
- values(somecolumn="x").
- with_hint("XYZ", "mysql"),
+ t.update().where(t.c.somecolumn == "q").
+ values(somecolumn="x").
+ with_hint("XYZ", "mysql"),
"UPDATE sometable SET somecolumn=:somecolumn "
"WHERE sometable.somecolumn = :somecolumn_1"
)
for targ in (None, t):
for darg in ("*", "mssql"):
self.assert_compile(
- t.delete().where(t.c.somecolumn=="q").
- with_hint("WITH (PAGLOCK)",
- selectable=targ,
- dialect_name=darg),
+ t.delete().where(t.c.somecolumn == "q").
+ with_hint("WITH (PAGLOCK)",
+ selectable=targ,
+ dialect_name=darg),
"DELETE FROM sometable WITH (PAGLOCK) "
"WHERE sometable.somecolumn = :somecolumn_1"
)
def test_delete_exclude_hint(self):
t = table('sometable', column('somecolumn'))
self.assert_compile(
- t.delete().\
- where(t.c.somecolumn=="q").\
- with_hint("XYZ", dialect_name="mysql"),
+ t.delete().
+ where(t.c.somecolumn == "q").
+ with_hint("XYZ", dialect_name="mysql"),
"DELETE FROM sometable WHERE "
"sometable.somecolumn = :somecolumn_1"
)
t2 = table('othertable', column('somecolumn'))
for darg in ("*", "mssql"):
self.assert_compile(
- t.update().where(t.c.somecolumn==t2.c.somecolumn).
- values(somecolumn="x").
- with_hint("WITH (PAGLOCK)",
- selectable=t2,
- dialect_name=darg),
+ t.update().where(t.c.somecolumn == t2.c.somecolumn).
+ values(somecolumn="x").
+ with_hint("WITH (PAGLOCK)",
+ selectable=t2,
+ dialect_name=darg),
"UPDATE sometable SET somecolumn=:somecolumn "
"FROM sometable, othertable WITH (PAGLOCK) "
"WHERE sometable.somecolumn = othertable.somecolumn"
)
# TODO: not supported yet.
- #def test_delete_from_hint(self):
+ # def test_delete_from_hint(self):
# t = table('sometable', column('somecolumn'))
# t2 = table('othertable', column('somecolumn'))
# for darg in ("*", "mssql"):
"IN ('x', 'y', 'z')",
),
(
- t.c.foo.in_([None]),
- "sometable.foo IN (NULL)"
+ t.c.foo.in_([None]),
+ "sometable.foo IN (NULL)"
)
]:
self.assert_compile(expr, compile, dialect=mxodbc_dialect)
t = table('sometable', column('somecolumn'))
self.assert_compile(t.select().where(t.c.somecolumn
- == t.select()),
+ == t.select()),
'SELECT sometable.somecolumn FROM '
'sometable WHERE sometable.somecolumn = '
'(SELECT sometable.somecolumn FROM '
'sometable)')
self.assert_compile(t.select().where(t.c.somecolumn
- != t.select()),
+ != t.select()),
'SELECT sometable.somecolumn FROM '
'sometable WHERE sometable.somecolumn != '
'(SELECT sometable.somecolumn FROM '
subqueries"""
table1 = table('mytable',
- column('myid', Integer),
- column('name', String),
- column('description', String),
- )
+ column('myid', Integer),
+ column('name', String),
+ column('description', String),
+ )
q = select([table1.c.myid],
order_by=[table1.c.myid]).alias('foo')
"myid FROM mytable) AS foo, mytable WHERE "
"foo.myid = mytable.myid")
-
-
def test_delete_schema(self):
metadata = MetaData()
tbl = Table('test', metadata, Column('id', Integer,
- primary_key=True), schema='paj')
+ primary_key=True), schema='paj')
self.assert_compile(tbl.delete(tbl.c.id == 1),
'DELETE FROM paj.test WHERE paj.test.id = '
':id_1')
def test_delete_schema_multipart(self):
metadata = MetaData()
- tbl = Table('test', metadata, Column('id', Integer,
- primary_key=True), schema='banana.paj')
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer,
+ primary_key=True),
+ schema='banana.paj')
self.assert_compile(tbl.delete(tbl.c.id == 1),
'DELETE FROM banana.paj.test WHERE '
'banana.paj.test.id = :id_1')
def test_delete_schema_multipart_needs_quoting(self):
metadata = MetaData()
- tbl = Table('test', metadata, Column('id', Integer,
- primary_key=True), schema='banana split.paj')
+ tbl = Table(
+ 'test', metadata,
+ Column('id', Integer, primary_key=True),
+ schema='banana split.paj')
self.assert_compile(tbl.delete(tbl.c.id == 1),
'DELETE FROM [banana split].paj.test WHERE '
'[banana split].paj.test.id = :id_1')
def test_delete_schema_multipart_both_need_quoting(self):
metadata = MetaData()
tbl = Table('test', metadata, Column('id', Integer,
- primary_key=True),
+ primary_key=True),
schema='banana split.paj with a space')
self.assert_compile(tbl.delete(tbl.c.id == 1),
'DELETE FROM [banana split].[paj with a '
':id_1)')
def test_union(self):
- t1 = table('t1', column('col1'), column('col2'), column('col3'
- ), column('col4'))
- t2 = table('t2', column('col1'), column('col2'), column('col3'
- ), column('col4'))
- s1, s2 = select([t1.c.col3.label('col3'), t1.c.col4.label('col4'
- )], t1.c.col2.in_(['t1col2r1', 't1col2r2'])), \
+ t1 = table(
+ 't1', column('col1'), column('col2'),
+ column('col3'), column('col4'))
+ t2 = table(
+ 't2', column('col1'), column('col2'),
+ column('col3'), column('col4'))
+ s1, s2 = select(
+ [t1.c.col3.label('col3'), t1.c.col4.label('col4')],
+ t1.c.col2.in_(['t1col2r1', 't1col2r2'])), \
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
t2.c.col2.in_(['t2col2r2', 't2col2r3']))
u = union(s1, s2, order_by=['col3', 'col4'])
self.assert_compile(func.current_time(), 'CURRENT_TIME')
self.assert_compile(func.foo(), 'foo()')
m = MetaData()
- t = Table('sometable', m, Column('col1', Integer), Column('col2'
- , Integer))
+ t = Table(
+ 'sometable', m, Column('col1', Integer), Column('col2', Integer))
self.assert_compile(select([func.max(t.c.col1)]),
'SELECT max(sometable.col1) AS max_1 FROM '
'sometable')
'SELECT DATEPART("%s", t.col1) AS anon_1 FROM t' % field)
def test_update_returning(self):
- table1 = table('mytable', column('myid', Integer), column('name'
- , String(128)), column('description',
- String(128)))
- u = update(table1, values=dict(name='foo'
- )).returning(table1.c.myid, table1.c.name)
+ table1 = table(
+ 'mytable',
+ column('myid', Integer),
+ column('name', String(128)),
+ column('description', String(128)))
+ u = update(
+ table1,
+ values=dict(name='foo')).returning(table1.c.myid, table1.c.name)
self.assert_compile(u,
'UPDATE mytable SET name=:name OUTPUT '
'inserted.myid, inserted.name')
'UPDATE mytable SET name=:name OUTPUT '
'inserted.myid, inserted.name, '
'inserted.description')
- u = update(table1, values=dict(name='foo'
- )).returning(table1).where(table1.c.name == 'bar')
+ u = update(
+ table1,
+ values=dict(
+ name='foo')).returning(table1).where(table1.c.name == 'bar')
self.assert_compile(u,
'UPDATE mytable SET name=:name OUTPUT '
'inserted.myid, inserted.name, '
'inserted.description WHERE mytable.name = '
':name_1')
u = update(table1, values=dict(name='foo'
- )).returning(func.length(table1.c.name))
+ )).returning(func.length(table1.c.name))
self.assert_compile(u,
'UPDATE mytable SET name=:name OUTPUT '
'LEN(inserted.name) AS length_1')
def test_delete_returning(self):
- table1 = table('mytable', column('myid', Integer), column('name'
- , String(128)), column('description',
- String(128)))
+ table1 = table(
+ 'mytable', column('myid', Integer),
+ column('name', String(128)), column('description', String(128)))
d = delete(table1).returning(table1.c.myid, table1.c.name)
self.assert_compile(d,
'DELETE FROM mytable OUTPUT deleted.myid, '
'deleted.name')
d = delete(table1).where(table1.c.name == 'bar'
).returning(table1.c.myid,
- table1.c.name)
+ table1.c.name)
self.assert_compile(d,
'DELETE FROM mytable OUTPUT deleted.myid, '
'deleted.name WHERE mytable.name = :name_1')
def test_insert_returning(self):
- table1 = table('mytable', column('myid', Integer), column('name'
- , String(128)), column('description',
- String(128)))
- i = insert(table1, values=dict(name='foo'
- )).returning(table1.c.myid, table1.c.name)
+ table1 = table(
+ 'mytable', column('myid', Integer),
+ column('name', String(128)), column('description', String(128)))
+ i = insert(
+ table1,
+ values=dict(name='foo')).returning(table1.c.myid, table1.c.name)
self.assert_compile(i,
'INSERT INTO mytable (name) OUTPUT '
'inserted.myid, inserted.name VALUES '
'inserted.myid, inserted.name, '
'inserted.description VALUES (:name)')
i = insert(table1, values=dict(name='foo'
- )).returning(func.length(table1.c.name))
+ )).returning(func.length(table1.c.name))
self.assert_compile(i,
'INSERT INTO mytable (name) OUTPUT '
'LEN(inserted.name) AS length_1 VALUES '
def test_limit_using_top(self):
t = table('t', column('x', Integer), column('y', Integer))
- s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(10)
+ s = select([t]).where(t.c.x == 5).order_by(t.c.y).limit(10)
self.assert_compile(
s,
def test_limit_zero_using_top(self):
t = table('t', column('x', Integer), column('y', Integer))
- s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(0)
+ s = select([t]).where(t.c.x == 5).order_by(t.c.y).limit(0)
self.assert_compile(
s,
def test_limit_offset_using_window(self):
t = table('t', column('x', Integer), column('y', Integer))
- s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(10).offset(20)
+ s = select([t]).where(t.c.x == 5).order_by(t.c.y).limit(10).offset(20)
self.assert_compile(
s,
def test_limit_zero_offset_using_window(self):
t = table('t', column('x', Integer), column('y', Integer))
- s = select([t]).where(t.c.x==5).order_by(t.c.y).limit(0).offset(0)
+ s = select([t]).where(t.c.x == 5).order_by(t.c.y).limit(0).offset(0)
# render the LIMIT of zero, but not the OFFSET
# of zero, so produces TOP 0
metadata = MetaData()
tbl = Table('test', metadata,
Column('id', Integer, Sequence('', 0), primary_key=True))
- self.assert_compile(schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
- "PRIMARY KEY (id))"
- )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
+ "PRIMARY KEY (id))"
+ )
def test_sequence_non_primary_key(self):
metadata = MetaData()
tbl = Table('test', metadata,
Column('id', Integer, Sequence(''), primary_key=False))
- self.assert_compile(schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))"
- )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))"
+ )
def test_sequence_ignore_nullability(self):
metadata = MetaData()
tbl = Table('test', metadata,
Column('id', Integer, Sequence(''), nullable=True))
- self.assert_compile(schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))"
- )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))"
+ )
def test_table_pkc_clustering(self):
metadata = MetaData()
Column('x', Integer, autoincrement=False),
Column('y', Integer, autoincrement=False),
PrimaryKeyConstraint("x", "y", mssql_clustered=True))
- self.assert_compile(schema.CreateTable(tbl),
- "CREATE TABLE test (x INTEGER NOT NULL, y INTEGER NOT NULL, "
- "PRIMARY KEY CLUSTERED (x, y))"
- )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (x INTEGER NOT NULL, y INTEGER NOT NULL, "
+ "PRIMARY KEY CLUSTERED (x, y))"
+ )
def test_table_uc_clustering(self):
metadata = MetaData()
Column('y', Integer, autoincrement=False),
PrimaryKeyConstraint("x"),
UniqueConstraint("y", mssql_clustered=True))
- self.assert_compile(schema.CreateTable(tbl),
- "CREATE TABLE test (x INTEGER NOT NULL, y INTEGER NULL, "
- "PRIMARY KEY (x), UNIQUE CLUSTERED (y))"
- )
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE test (x INTEGER NOT NULL, y INTEGER NULL, "
+ "PRIMARY KEY (x), UNIQUE CLUSTERED (y))"
+ )
def test_index_clustering(self):
metadata = MetaData()
def test_index_ordering(self):
metadata = MetaData()
- tbl = Table('test', metadata,
- Column('x', Integer), Column('y', Integer), Column('z', Integer))
+ tbl = Table(
+ 'test', metadata,
+ Column('x', Integer), Column('y', Integer), Column('z', Integer))
idx = Index("foo", tbl.c.x.desc(), "y")
self.assert_compile(schema.CreateIndex(idx),
"CREATE INDEX foo ON test (x DESC, y)"
def test_create_index_expr(self):
m = MetaData()
t1 = Table('foo', m,
- Column('x', Integer)
- )
+ Column('x', Integer)
+ )
self.assert_compile(
schema.CreateIndex(Index("bar", t1.c.x > 5)),
"CREATE INDEX bar ON foo (x > 5)"
def test_drop_index_w_schema(self):
m = MetaData()
t1 = Table('foo', m,
- Column('x', Integer),
- schema='bar'
- )
+ Column('x', Integer),
+ schema='bar'
+ )
self.assert_compile(
schema.DropIndex(Index("idx_foo", t1.c.x)),
"DROP INDEX idx_foo ON bar.foo"
def test_index_extra_include_1(self):
metadata = MetaData()
- tbl = Table('test', metadata,
- Column('x', Integer), Column('y', Integer), Column('z', Integer))
+ tbl = Table(
+ 'test', metadata,
+ Column('x', Integer), Column('y', Integer), Column('z', Integer))
idx = Index("foo", tbl.c.x, mssql_include=['y'])
self.assert_compile(schema.CreateIndex(idx),
"CREATE INDEX foo ON test (x) INCLUDE (y)"
def test_index_extra_include_2(self):
metadata = MetaData()
- tbl = Table('test', metadata,
- Column('x', Integer), Column('y', Integer), Column('z', Integer))
+ tbl = Table(
+ 'test', metadata,
+ Column('x', Integer), Column('y', Integer), Column('z', Integer))
idx = Index("foo", tbl.c.x, mssql_include=[tbl.c.y])
self.assert_compile(schema.CreateIndex(idx),
"CREATE INDEX foo ON test (x) INCLUDE (y)"
def setup(self):
t = Table('sometable', MetaData(),
- Column('pk_column', Integer),
- Column('test_column', String)
- )
+ Column('pk_column', Integer),
+ Column('test_column', String)
+ )
self.column = t.c.test_column
dialect = mssql.dialect()
self.ddl_compiler = dialect.ddl_compiler(dialect,
- schema.CreateTable(t))
+ schema.CreateTable(t))
def _column_spec(self):
return self.ddl_compiler.get_column_specification(self.column)
# -*- encoding: utf-8
from sqlalchemy.testing import eq_, engines
-from sqlalchemy import *
from sqlalchemy.sql import table, column
from sqlalchemy.databases import mssql
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from sqlalchemy.util import ue
from sqlalchemy import util
from sqlalchemy.testing.assertsql import CursorSQL
+from sqlalchemy import Integer, String, Table, Column, select, MetaData,\
+ func, PrimaryKeyConstraint, desc, Sequence, DDL, ForeignKey, or_, and_
+from sqlalchemy import event
+metadata = None
+cattable = None
+matchtable = None
class SchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL):
- """SQL server cannot reference schema-qualified tables in a SELECT statement, they
- must be aliased.
+
+ """SQL server cannot reference schema-qualified tables in a SELECT
+ statement, they must be aliased.
"""
__dialect__ = mssql.dialect()
def setup(self):
metadata = MetaData()
self.t1 = table('t1',
- column('a', Integer),
- column('b', String),
- column('c', String),
- )
+ column('a', Integer),
+ column('b', String),
+ column('c', String),
+ )
self.t2 = Table(
't2', metadata,
Column("a", Integer),
Column("b", Integer),
Column("c", Integer),
- schema = 'schema'
+ schema='schema'
)
def test_result_map(self):
assert self.t2.c.a in set(c._create_result_map()['schema_t2_a'][1])
def test_straight_select(self):
- self.assert_compile(self.t2.select(),
+ self.assert_compile(
+ self.t2.select(),
"SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1"
)
def test_join_to_schema(self):
t1, t2 = self.t1, self.t2
self.assert_compile(
- t1.join(t2, t1.c.a==t2.c.a).select(),
+ t1.join(t2, t1.c.a == t2.c.a).select(),
"SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 "
"JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a"
)
def test_union_schema_to_non(self):
t1, t2 = self.t1, self.t2
s = select([t2.c.a, t2.c.b]).apply_labels().\
- union(
- select([t1.c.a, t1.c.b]).apply_labels()
- ).alias().select()
+ union(
+ select([t1.c.a, t1.c.b]).apply_labels()
+ ).alias().select()
self.assert_compile(
s,
"SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
)
+
class IdentityInsertTest(fixtures.TestBase, AssertsCompiledSQL):
__only_on__ = 'mssql'
__dialect__ = mssql.MSDialect()
metadata = MetaData(testing.db)
cattable = Table('cattable', metadata,
- Column('id', Integer),
- Column('description', String(50)),
- PrimaryKeyConstraint('id', name='PK_cattable'),
- )
+ Column('id', Integer),
+ Column('description', String(50)),
+ PrimaryKeyConstraint('id', name='PK_cattable'),
+ )
def setup(self):
metadata.create_all()
def test_compiled(self):
self.assert_compile(cattable.insert().values(id=9,
- description='Python'),
+ description='Python'),
'INSERT INTO cattable (id, description) '
'VALUES (:id, :description)')
def test_executemany(self):
cattable.insert().execute([{'id': 89, 'description': 'Python'},
- {'id': 8, 'description': 'Ruby'},
- {'id': 3, 'description': 'Perl'},
- {'id': 1, 'description': 'Java'}])
+ {'id': 8, 'description': 'Ruby'},
+ {'id': 3, 'description': 'Perl'},
+ {'id': 1, 'description': 'Java'}])
cats = cattable.select().order_by(cattable.c.id).execute()
eq_([(1, 'Java'), (3, 'Perl'), (8, 'Ruby'), (89, 'Python')],
list(cats))
cattable.insert().execute([{'description': 'PHP'},
- {'description': 'Smalltalk'}])
+ {'description': 'Smalltalk'}])
lastcats = \
cattable.select().order_by(desc(cattable.c.id)).limit(2).execute()
eq_([(91, 'Smalltalk'), (90, 'PHP')], list(lastcats))
+
class QueryUnicodeTest(fixtures.TestBase):
__only_on__ = 'mssql'
def test_convert_unicode(self):
meta = MetaData(testing.db)
- t1 = Table('unitest_table', meta, Column('id', Integer,
- primary_key=True), Column('descr',
- mssql.MSText(convert_unicode=True)))
+ t1 = Table(
+ 'unitest_table', meta,
+ Column('id', Integer, primary_key=True),
+ Column('descr', mssql.MSText(convert_unicode=True)))
meta.create_all()
con = testing.db.connect()
try:
r = t1.select().execute().first()
assert isinstance(r[1], util.text_type), \
- '%s is %s instead of unicode, working on %s' % (r[1],
- type(r[1]), meta.bind)
+ '%s is %s instead of unicode, working on %s' % (
+ r[1],
+ type(r[1]), meta.bind)
finally:
meta.drop_all()
+
class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
__only_on__ = 'mssql'
with the init parameter 'implicit_returning = False'.
"""
- #todo: this same test needs to be tried in a multithreaded context
+ # todo: this same test needs to be tried in a multithreaded context
# with multiple threads inserting to the same table.
- #todo: check whether this error also occurs with clients other
+ # todo: check whether this error also occurs with clients other
# than the SQL Server Native Client. Maybe an assert_raises
# test should be written.
meta = MetaData(testing.db)
t1 = Table('t1', meta,
- Column('id', Integer, Sequence('fred', 100, 1),
- primary_key=True),
- Column('descr', String(200)),
- # the following flag will prevent the
- # MSSQLCompiler.returning_clause from getting called,
- # though the ExecutionContext will still have a
- # _select_lastrowid, so the SELECT SCOPE_IDENTITY() will
- # hopefully be called instead.
- implicit_returning = False
- )
+ Column('id', Integer, Sequence('fred', 100, 1),
+ primary_key=True),
+ Column('descr', String(200)),
+ # the following flag will prevent the
+ # MSSQLCompiler.returning_clause from getting called,
+ # though the ExecutionContext will still have a
+ # _select_lastrowid, so the SELECT SCOPE_IDENTITY() will
+ # hopefully be called instead.
+ implicit_returning=False
+ )
t2 = Table('t2', meta,
- Column('id', Integer, Sequence('fred', 200, 1),
- primary_key=True),
- Column('descr', String(200)))
+ Column('id', Integer, Sequence('fred', 200, 1),
+ primary_key=True),
+ Column('descr', String(200)))
meta.create_all()
con = testing.db.connect()
con.execute("""create trigger paj on t1 for insert as
),
)
+ @testing.provide_metadata
def test_insertid_schema(self):
- meta = MetaData(testing.db)
+ meta = self.metadata
con = testing.db.connect()
con.execute('create schema paj')
+
+ @event.listens_for(meta, "after_drop")
+ def cleanup(target, connection, **kw):
+ connection.execute('drop schema paj')
+
tbl = Table('test', meta,
Column('id', Integer, primary_key=True), schema='paj')
tbl.create()
- try:
- tbl.insert().execute({'id':1})
- finally:
- tbl.drop()
- con.execute('drop schema paj')
+ tbl.insert().execute({'id': 1})
+ eq_(tbl.select().scalar(), 1)
+ @testing.provide_metadata
def test_returning_no_autoinc(self):
- meta = MetaData(testing.db)
- table = Table('t1', meta, Column('id', Integer,
- primary_key=True), Column('data', String(50)))
+ meta = self.metadata
+ table = Table(
+ 't1', meta,
+ Column('id', Integer, primary_key=True),
+ Column('data', String(50)))
table.create()
- try:
- result = table.insert().values(id=1,
- data=func.lower('SomeString'
- )).returning(table.c.id, table.c.data).execute()
- eq_(result.fetchall(), [(1, 'somestring')])
- finally:
-
- # this will hang if the "SET IDENTITY_INSERT t1 OFF" occurs
- # before the result is fetched
-
- table.drop()
+ result = table.insert().values(
+ id=1,
+ data=func.lower('SomeString')).\
+ returning(table.c.id, table.c.data).execute()
+ eq_(result.fetchall(), [(1, 'somestring')])
+ @testing.provide_metadata
def test_delete_schema(self):
- meta = MetaData(testing.db)
+ meta = self.metadata
con = testing.db.connect()
con.execute('create schema paj')
- tbl = Table('test', meta, Column('id', Integer,
- primary_key=True), schema='paj')
+
+ @event.listens_for(meta, "after_drop")
+ def cleanup(target, connection, **kw):
+ connection.execute('drop schema paj')
+
+ tbl = Table(
+ 'test', meta,
+ Column('id', Integer, primary_key=True), schema='paj')
tbl.create()
- try:
- tbl.insert().execute({'id': 1})
- tbl.delete(tbl.c.id == 1).execute()
- finally:
- tbl.drop()
- con.execute('drop schema paj')
+ tbl.insert().execute({'id': 1})
+ eq_(tbl.select().scalar(), 1)
+ tbl.delete(tbl.c.id == 1).execute()
+ eq_(tbl.select().scalar(), None)
+ @testing.provide_metadata
def test_insertid_reserved(self):
- meta = MetaData(testing.db)
+ meta = self.metadata
table = Table(
'select', meta,
Column('col', Integer, primary_key=True)
)
table.create()
- meta2 = MetaData(testing.db)
- try:
- table.insert().execute(col=7)
- finally:
- table.drop()
+ table.insert().execute(col=7)
+ eq_(table.select().scalar(), 7)
class Foo(object):
+
def __init__(self, **kw):
for k in kw:
setattr(self, k, kw[k])
finally:
connection.close()
+
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
__only_on__ = 'mssql'
Column('title', String(200)),
Column('category_id', Integer, ForeignKey('cattable.id')),
PrimaryKeyConstraint('id', name='PK_matchtable'),
- )
+ )
DDL("""CREATE FULLTEXT INDEX
ON cattable (description)
- KEY INDEX PK_cattable""").execute_at('after-create'
- , matchtable)
+ KEY INDEX PK_cattable""").\
+ execute_at('after-create', matchtable)
DDL("""CREATE FULLTEXT INDEX
ON matchtable (title)
- KEY INDEX PK_matchtable""").execute_at('after-create'
- , matchtable)
+ KEY INDEX PK_matchtable""").\
+ execute_at('after-create', matchtable)
metadata.create_all()
cattable.insert().execute([{'id': 1, 'description': 'Python'},
- {'id': 2, 'description': 'Ruby'}])
- matchtable.insert().execute([{'id': 1, 'title'
- : 'Agile Web Development with Rails'
- , 'category_id': 2}, {'id': 2,
- 'title': 'Dive Into Python',
- 'category_id': 1}, {'id': 3, 'title'
- : "Programming Matz's Ruby",
- 'category_id': 2}, {'id': 4, 'title'
- : 'The Definitive Guide to Django',
- 'category_id': 1}, {'id': 5, 'title'
- : 'Python in a Nutshell',
- 'category_id': 1}])
+ {'id': 2, 'description': 'Ruby'}])
+ matchtable.insert().execute([
+ {'id': 1, 'title': 'Web Development with Rails', 'category_id': 2},
+ {'id': 2, 'title': 'Dive Into Python', 'category_id': 1},
+ {'id': 3, 'title': "Programming Matz's Ruby", 'category_id': 2},
+ {'id': 4, 'title': 'Guide to Django', 'category_id': 1},
+ {'id': 5, 'title': 'Python in a Nutshell', 'category_id': 1}])
DDL("WAITFOR DELAY '00:00:05'"
).execute(bind=engines.testing_engine())
def test_simple_match(self):
results = \
- matchtable.select().where(matchtable.c.title.match('python'
- )).order_by(matchtable.c.id).execute().fetchall()
+ matchtable.select().where(
+ matchtable.c.title.match('python')).\
+ order_by(matchtable.c.id).execute().fetchall()
eq_([2, 5], [r.id for r in results])
def test_simple_match_with_apostrophe(self):
results = \
- matchtable.select().where(matchtable.c.title.match("Matz's"
- )).execute().fetchall()
+ matchtable.select().where(
+ matchtable.c.title.match("Matz's")).execute().fetchall()
eq_([3], [r.id for r in results])
def test_simple_prefix_match(self):
results = \
- matchtable.select().where(matchtable.c.title.match('"nut*"'
- )).execute().fetchall()
+ matchtable.select().where(
+ matchtable.c.title.match('"nut*"')).execute().fetchall()
eq_([5], [r.id for r in results])
def test_simple_inflectional_match(self):
results = \
matchtable.select().where(
matchtable.c.title.match('FORMSOF(INFLECTIONAL, "dives")'
- )).execute().fetchall()
+ )).execute().fetchall()
eq_([2], [r.id for r in results])
def test_or_match(self):
results1 = \
- matchtable.select().where(or_(matchtable.c.title.match('nutshell'
- ), matchtable.c.title.match('ruby'
- ))).order_by(matchtable.c.id).execute().fetchall()
+ matchtable.select().where(or_(
+ matchtable.c.title.match('nutshell'),
+ matchtable.c.title.match('ruby'))).\
+ order_by(matchtable.c.id).execute().fetchall()
eq_([3, 5], [r.id for r in results1])
results2 = \
matchtable.select().where(
- matchtable.c.title.match('nutshell OR ruby'
- )).order_by(matchtable.c.id).execute().fetchall()
+ matchtable.c.title.match(
+ 'nutshell OR ruby')).\
+ order_by(matchtable.c.id).execute().fetchall()
eq_([3, 5], [r.id for r in results2])
def test_and_match(self):
results1 = \
- matchtable.select().where(and_(matchtable.c.title.match('python'
- ), matchtable.c.title.match('nutshell'
- ))).execute().fetchall()
+ matchtable.select().where(and_(
+ matchtable.c.title.match('python'),
+ matchtable.c.title.match('nutshell'))).execute().fetchall()
eq_([5], [r.id for r in results1])
results2 = \
matchtable.select().where(
matchtable.c.title.match('python AND nutshell'
- )).execute().fetchall()
+ )).execute().fetchall()
eq_([5], [r.id for r in results2])
def test_match_across_joins(self):
- results = matchtable.select().where(and_(cattable.c.id
- == matchtable.c.category_id,
- or_(cattable.c.description.match('Ruby'),
- matchtable.c.title.match('nutshell'
- )))).order_by(matchtable.c.id).execute().fetchall()
+ results = matchtable.select().where(
+ and_(cattable.c.id == matchtable.c.category_id,
+ or_(cattable.c.description.match('Ruby'),
+ matchtable.c.title.match('nutshell')))).\
+ order_by(matchtable.c.id).execute().fetchall()
eq_([1, 3, 5], [r.id for r in results])
-
-
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
