-from sqlalchemy import *
-from sqlalchemy.testing import *
from sqlalchemy.engine import default
+from sqlalchemy.testing import fixtures, AssertsExecutionResults, profiling
+from sqlalchemy import MetaData, Table, Column, Integer, String, select
+
+t1 = t2 = None
+
class CompileTest(fixtures.TestBase, AssertsExecutionResults):
__requires__ = 'cpython',
global t1, t2, metadata
metadata = MetaData()
t1 = Table('t1', metadata,
- Column('c1', Integer, primary_key=True),
- Column('c2', String(30)))
+ Column('c1', Integer, primary_key=True),
+ Column('c2', String(30)))
t2 = Table('t2', metadata,
- Column('c1', Integer, primary_key=True),
- Column('c2', String(30)))
+ Column('c1', Integer, primary_key=True),
+ Column('c2', String(30)))
# do a "compile" ahead of time to load
# deferred imports
def go():
s = select([t1], t1.c.c2 == t2.c.c1).apply_labels()
s.compile(dialect=self.dialect)
- go()
\ No newline at end of file
+ go()
from sqlalchemy.testing import fixtures
import weakref
+
class A(fixtures.ComparableEntity):
pass
+
+
class B(fixtures.ComparableEntity):
pass
+
+
class ASub(A):
pass
+
def profile_memory(maxtimes=50):
def decorate(func):
# run the test N times. if length of gc.get_objects()
return profile
return decorate
+
def assert_no_mappers():
clear_mappers()
gc_collect()
assert len(_mapper_registry) == 0
+
class EnsureZeroed(fixtures.ORMTest):
+
def setup(self):
_sessions.clear()
_mapper_registry.clear()
+
class MemUsageTest(EnsureZeroed):
__requires__ = 'cpython',
pass
x = []
+
@profile_memory(maxtimes=10)
def go():
x[-1:] = [Foo(), Foo(), Foo(), Foo(), Foo(), Foo()]
metadata = MetaData(testing.db)
table1 = Table("mytable", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)))
table2 = Table("mytable2", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)),
- Column('col3', Integer, ForeignKey("mytable.col1")))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)),
+ Column('col3', Integer, ForeignKey("mytable.col1")))
metadata.create_all()
m1 = mapper(A, table1, properties={
- "bs":relationship(B, cascade="all, delete",
- order_by=table2.c.col1)},
- order_by=table1.c.col1)
+ "bs": relationship(B, cascade="all, delete",
+ order_by=table2.c.col1)},
+ order_by=table1.c.col1)
m2 = mapper(B, table2)
m3 = mapper(A, table1, non_primary=True)
a1.bs.append(B(col2="b1"))
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
- for x in [a1,a2,a3]:
+ for x in [a1, a2, a3]:
sess.add(x)
sess.flush()
sess.expunge_all()
metadata = MetaData(testing.db)
table1 = Table("mytable", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)))
table2 = Table("mytable2", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)),
- Column('col3', Integer, ForeignKey("mytable.col1")))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)),
+ Column('col3', Integer, ForeignKey("mytable.col1")))
metadata.create_all()
m1 = mapper(A, table1, properties={
- "bs":relationship(B, cascade="all, delete",
- order_by=table2.c.col1)},
- order_by=table1.c.col1,
- _compiled_cache_size=10
- )
+ "bs": relationship(B, cascade="all, delete",
+ order_by=table2.c.col1)},
+ order_by=table1.c.col1,
+ _compiled_cache_size=10
+ )
m2 = mapper(B, table2,
- _compiled_cache_size=10
- )
+ _compiled_cache_size=10
+ )
m3 = mapper(A, table1, non_primary=True)
@profile_memory()
def go():
engine = engines.testing_engine(
- options={'logging_name':'FOO',
- 'pool_logging_name':'BAR',
- 'use_reaper':False}
- )
+ options={'logging_name': 'FOO',
+ 'pool_logging_name': 'BAR',
+ 'use_reaper': False}
+ )
sess = create_session(bind=engine)
a1 = A(col2="a1")
a1.bs.append(B(col2="b1"))
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
- for x in [a1,a2,a3]:
+ for x in [a1, a2, a3]:
sess.add(x)
sess.flush()
sess.expunge_all()
type_ = args[0](*args[1:])
bp = type_._cached_bind_processor(eng.dialect)
rp = type_._cached_result_processor(eng.dialect, 0)
+ bp, rp # strong reference
go()
assert not eng.dialect._type_memos
-
def test_many_updates(self):
metadata = MetaData(testing.db)
wide_table = Table('t', metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- *[Column('col%d' % i, Integer) for i in range(10)]
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ *[Column('col%d' % i, Integer) for i in range(10)]
+ )
class Wide(object):
pass
@testing.crashes('mysql+cymysql', 'blocking')
def test_unicode_warnings(self):
metadata = MetaData(testing.db)
- table1 = Table('mytable', metadata, Column('col1', Integer,
- primary_key=True,
- test_needs_autoincrement=True), Column('col2',
- Unicode(30)))
+ table1 = Table(
+ 'mytable',
+ metadata,
+ Column(
+ 'col1',
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'col2',
+ Unicode(30)))
metadata.create_all()
i = [1]
# this warning shouldn't clog up memory.
testing.db.execute(table1.select().where(table1.c.col2
- == 'foo%d' % i[0]))
+ == 'foo%d' % i[0]))
i[0] += 1
try:
go()
metadata = MetaData(testing.db)
table1 = Table("mytable", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)))
table2 = Table("mytable2", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)),
- Column('col3', Integer, ForeignKey("mytable.col1")))
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)),
+ Column('col3', Integer, ForeignKey("mytable.col1")))
@profile_memory()
def go():
- m1 = mapper(A, table1, properties={
- "bs":relationship(B, order_by=table2.c.col1)
+ mapper(A, table1, properties={
+ "bs": relationship(B, order_by=table2.c.col1)
})
- m2 = mapper(B, table2)
+ mapper(B, table2)
- m3 = mapper(A, table1, non_primary=True)
+ mapper(A, table1, non_primary=True)
sess = create_session()
a1 = A(col2="a1")
a1.bs.append(B(col2="b1"))
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
- for x in [a1,a2,a3]:
+ for x in [a1, a2, a3]:
sess.add(x)
sess.flush()
sess.expunge_all()
metadata = MetaData(testing.db)
a = Table("a", metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('bid', Integer, ForeignKey('b.id')),
- Column('type', String(30))
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('bid', Integer, ForeignKey('b.id')),
+ Column('type', String(30))
+ )
asub = Table("asub", metadata,
- Column('id', Integer, ForeignKey('a.id'),
- primary_key=True),
- Column('data', String(30)))
+ Column('id', Integer, ForeignKey('a.id'),
+ primary_key=True),
+ Column('data', String(30)))
b = Table("b", metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- )
+ Column('id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ )
mapper(A, a, polymorphic_identity='a',
- polymorphic_on=a.c.type)
- mapper(ASub, asub, inherits=A,polymorphic_identity='asub')
- m1 = mapper(B, b, properties={
- 'as_':relationship(A)
+ polymorphic_on=a.c.type)
+ mapper(ASub, asub, inherits=A, polymorphic_identity='asub')
+ mapper(B, b, properties={
+ 'as_': relationship(A)
})
metadata.create_all()
def test_path_registry(self):
metadata = MetaData()
a = Table("a", metadata,
- Column('id', Integer, primary_key=True),
- Column('foo', Integer),
- Column('bar', Integer)
- )
+ Column('id', Integer, primary_key=True),
+ Column('foo', Integer),
+ Column('bar', Integer)
+ )
m1 = mapper(A, a)
+
@profile_memory()
def go():
ma = sa.inspect(aliased(A))
metadata = MetaData(testing.db)
table1 = Table("mytable", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30))
- )
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30))
+ )
table2 = Table("mytable2", metadata,
- Column('col1', Integer, ForeignKey('mytable.col1'),
- primary_key=True, test_needs_autoincrement=True),
- Column('col3', String(30)),
- )
+ Column('col1', Integer, ForeignKey('mytable.col1'),
+ primary_key=True, test_needs_autoincrement=True),
+ Column('col3', String(30)),
+ )
@profile_memory()
def go():
class A(fixtures.ComparableEntity):
pass
+
class B(A):
pass
a2 = A()
b1 = B(col3='b1')
b2 = B(col3='b2')
- for x in [a1,a2,b1, b2]:
+ for x in [a1, a2, b1, b2]:
sess.add(x)
sess.flush()
sess.expunge_all()
metadata = MetaData(testing.db)
table1 = Table("mytable", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30))
- )
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30))
+ )
table2 = Table("mytable2", metadata,
- Column('col1', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('col2', String(30)),
- )
+ Column('col1', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column('col2', String(30)),
+ )
table3 = Table('t1tot2', metadata,
- Column('t1', Integer, ForeignKey('mytable.col1')),
- Column('t2', Integer, ForeignKey('mytable2.col1')),
- )
+ Column('t1', Integer, ForeignKey('mytable.col1')),
+ Column('t2', Integer, ForeignKey('mytable2.col1')),
+ )
@profile_memory()
def go():
class A(fixtures.ComparableEntity):
pass
+
class B(fixtures.ComparableEntity):
pass
mapper(A, table1, properties={
- 'bs':relationship(B, secondary=table3,
- backref='as', order_by=table3.c.t1)
+ 'bs': relationship(B, secondary=table3,
+ backref='as', order_by=table3.c.t1)
})
mapper(B, table2)
b2 = B(col2='b2')
a1.bs.append(b1)
a2.bs.append(b2)
- for x in [a1,a2]:
+ for x in [a1, a2]:
sess.add(x)
sess.flush()
sess.expunge_all()
m = self.metadata
t = Table('t', m, Column('x', Integer), Column('y', Integer))
m.create_all(e)
- e.execute(t.insert(), {"x":1, "y":1})
+ e.execute(t.insert(), {"x": 1, "y": 1})
+
@profile_memory()
def go():
r = e.execute(t.alias().select())
@testing.crashes('mysql+cymysql', 'blocking')
def test_join_cache(self):
metadata = MetaData(testing.db)
- table1 = Table('table1', metadata, Column('id', Integer,
- primary_key=True,
- test_needs_autoincrement=True), Column('data',
- String(30)))
- table2 = Table('table2', metadata, Column('id', Integer,
- primary_key=True,
- test_needs_autoincrement=True), Column('data',
- String(30)), Column('t1id', Integer,
- ForeignKey('table1.id')))
+ table1 = Table(
+ 'table1',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data',
+ String(30)))
+ table2 = Table(
+ 'table2', metadata,
+ Column(
+ 'id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data', String(30)), Column(
+ 't1id', Integer, ForeignKey('table1.id')))
class Foo(object):
pass
class Bar(object):
pass
- mapper(Foo, table1, properties={'bars'
- : relationship(mapper(Bar, table2))})
+ mapper(Foo, table1, properties={
+ 'bars': relationship(mapper(Bar, table2))})
metadata.create_all()
session = sessionmaker()
finally:
metadata.drop_all()
-
-
def test_type_compile(self):
from sqlalchemy.dialects.sqlite.base import dialect as SQLiteDialect
cast = sa.cast(column('x'), sa.Integer)
+
@profile_memory()
def go():
dialect = SQLiteDialect()
def go():
to_unicode_processor_factory('utf8')
go()
-
-
-from sqlalchemy.testing import eq_, assert_raises, \
- assert_raises_message
-from sqlalchemy import exc as sa_exc, util, Integer, String, ForeignKey
-from sqlalchemy.orm import exc as orm_exc, mapper, relationship, \
+from sqlalchemy import exc as Integer, String, ForeignKey
+from sqlalchemy.orm import exc as mapper, relationship, \
sessionmaker, Session, defer
from sqlalchemy import testing
from sqlalchemy.testing import profiling
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.schema import Table, Column
-import sys
+
class MergeTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
- Table('parent', metadata, Column('id', Integer,
- primary_key=True,
- test_needs_autoincrement=True), Column('data',
- String(20)))
- Table('child', metadata, Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
- Column('data', String(20)), Column('parent_id',
- Integer, ForeignKey('parent.id'), nullable=False))
+ Table(
+ 'parent',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data',
+ String(20)))
+ Table(
+ 'child', metadata,
+ Column(
+ 'id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data', String(20)),
+ Column(
+ 'parent_id', Integer, ForeignKey('parent.id'), nullable=False))
@classmethod
def setup_classes(cls):
@classmethod
def setup_mappers(cls):
Child, Parent, parent, child = (cls.classes.Child,
- cls.classes.Parent,
- cls.tables.parent,
- cls.tables.child)
-
- mapper(Parent, parent, properties={'children':
- relationship(Child, backref='parent')})
+ cls.classes.Parent,
+ cls.tables.parent,
+ cls.tables.child)
+
+ mapper(
+ Parent,
+ parent,
+ properties={
+ 'children': relationship(
+ Child,
+ backref='parent')})
mapper(Child, child)
@classmethod
@profiling.function_call_count()
def go():
- p2 = sess2.merge(p1)
+ sess2.merge(p1)
go()
# one more time, count the SQL
def go2():
- p2 = sess2.merge(p1)
+ sess2.merge(p1)
sess2 = sessionmaker(testing.db)()
self.assert_sql_count(testing.db, go2, 2)
+
class LoadManyToOneFromIdentityTest(fixtures.MappedTest):
+
"""test overhead associated with many-to-one fetches.
Prior to the refactor of LoadLazyAttribute and
"""
-
@classmethod
def define_tables(cls, metadata):
Table('parent', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String(20)),
- Column('child_id', Integer, ForeignKey('child.id'))
- )
+ Column('id', Integer, primary_key=True),
+ Column('data', String(20)),
+ Column('child_id', Integer, ForeignKey('child.id'))
+ )
Table('child', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String(20))
- )
+ Column('id', Integer, primary_key=True),
+ Column('data', String(20))
+ )
@classmethod
def setup_classes(cls):
@classmethod
def setup_mappers(cls):
Child, Parent, parent, child = (cls.classes.Child,
- cls.classes.Parent,
- cls.tables.parent,
- cls.tables.child)
+ cls.classes.Parent,
+ cls.tables.parent,
+ cls.tables.child)
mapper(Parent, parent, properties={
'child': relationship(Child)})
parent, child = cls.tables.parent, cls.tables.child
child.insert().execute([
- {'id':i, 'data':'c%d' % i}
+ {'id': i, 'data': 'c%d' % i}
for i in range(1, 251)
])
parent.insert().execute([
{
- 'id':i,
- 'data':'p%dc%d' % (i, (i % 250) + 1),
- 'child_id':(i % 250) + 1
+ 'id': i,
+ 'data': 'p%dc%d' % (i, (i % 250) + 1),
+ 'child_id': (i % 250) + 1
}
for i in range(1, 1000)
])
sess = Session()
parents = sess.query(Parent).all()
-
@profiling.function_call_count(variance=.2)
def go():
for p in parents:
sess = Session()
parents = sess.query(Parent).all()
children = sess.query(Child).all()
+ children # strong reference
@profiling.function_call_count()
def go():
p.child
go()
+
class MergeBackrefsTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('a', metadata,
- Column('id', Integer, primary_key=True),
- Column('c_id', Integer, ForeignKey('c.id'))
- )
+ Column('id', Integer, primary_key=True),
+ Column('c_id', Integer, ForeignKey('c.id'))
+ )
Table('b', metadata,
- Column('id', Integer, primary_key=True),
- Column('a_id', Integer, ForeignKey('a.id'))
- )
+ Column('id', Integer, primary_key=True),
+ Column('a_id', Integer, ForeignKey('a.id'))
+ )
Table('c', metadata,
- Column('id', Integer, primary_key=True),
- )
+ Column('id', Integer, primary_key=True),
+ )
Table('d', metadata,
- Column('id', Integer, primary_key=True),
- Column('a_id', Integer, ForeignKey('a.id'))
- )
+ Column('id', Integer, primary_key=True),
+ Column('a_id', Integer, ForeignKey('a.id'))
+ )
@classmethod
def setup_classes(cls):
class A(cls.Basic):
pass
+
class B(cls.Basic):
pass
+
class C(cls.Basic):
pass
+
class D(cls.Basic):
pass
@classmethod
def setup_mappers(cls):
A, B, C, D = cls.classes.A, cls.classes.B, \
- cls.classes.C, cls.classes.D
+ cls.classes.C, cls.classes.D
a, b, c, d = cls.tables.a, cls.tables.b, \
- cls.tables.c, cls.tables.d
+ cls.tables.c, cls.tables.d
mapper(A, a, properties={
'bs': relationship(B, backref='a'),
'c': relationship(C, backref='as'),
@classmethod
def insert_data(cls):
A, B, C, D = cls.classes.A, cls.classes.B, \
- cls.classes.C, cls.classes.D
+ cls.classes.C, cls.classes.D
s = Session()
s.add_all([
A(id=i,
bs=[B(id=(i * 5) + j) for j in range(1, 5)],
c=C(id=i),
ds=[D(id=(i * 5) + j) for j in range(1, 5)]
- )
+ )
for i in range(1, 5)
])
s.commit()
@profiling.function_call_count(variance=.10)
def test_merge_pending_with_all_pks(self):
A, B, C, D = self.classes.A, self.classes.B, \
- self.classes.C, self.classes.D
+ self.classes.C, self.classes.D
s = Session()
for a in [
A(id=i,
bs=[B(id=(i * 5) + j) for j in range(1, 5)],
c=C(id=i),
ds=[D(id=(i * 5) + j) for j in range(1, 5)]
- )
+ )
for i in range(1, 5)
]:
s.merge(a)
+
class DeferOptionsTest(fixtures.MappedTest):
@classmethod
def define_tables(cls, metadata):
Table('a', metadata,
- Column('id', Integer, primary_key=True),
- Column('x', String(5)),
- Column('y', String(5)),
- Column('z', String(5)),
- Column('q', String(5)),
- Column('p', String(5)),
- Column('r', String(5)),
- )
+ Column('id', Integer, primary_key=True),
+ Column('x', String(5)),
+ Column('y', String(5)),
+ Column('z', String(5)),
+ Column('q', String(5)),
+ Column('p', String(5)),
+ Column('r', String(5)),
+ )
@classmethod
def setup_classes(cls):
s.add_all([
A(id=i,
**dict((letter, "%s%d" % (letter, i)) for letter in
- ['x', 'y', 'z', 'p', 'q', 'r'])
- ) for i in range(1, 1001)
+ ['x', 'y', 'z', 'p', 'q', 'r'])
+ ) for i in range(1, 1001)
])
s.commit()
@classmethod
def define_tables(cls, metadata):
- Table('parent', metadata, Column('id', Integer,
- primary_key=True,
- test_needs_autoincrement=True), Column('data',
- String(20)))
- Table('child', metadata, Column('id', Integer,
- primary_key=True, test_needs_autoincrement=True),
- Column('data', String(20)), Column('parent_id',
- Integer, ForeignKey('parent.id'), nullable=False))
+ Table(
+ 'parent',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data',
+ String(20)))
+ Table(
+ 'child', metadata,
+ Column(
+ 'id', Integer, primary_key=True,
+ test_needs_autoincrement=True),
+ Column(
+ 'data', String(20)), Column(
+ 'parent_id', Integer, ForeignKey('parent.id'), nullable=False))
@classmethod
def setup_classes(cls):
@classmethod
def setup_mappers(cls):
Child, Parent, parent, child = (cls.classes.Child,
- cls.classes.Parent,
- cls.tables.parent,
- cls.tables.child)
-
- mapper(Parent, parent, properties={'children':
- relationship(Child, backref='parent')})
+ cls.classes.Parent,
+ cls.tables.parent,
+ cls.tables.child)
+
+ mapper(
+ Parent,
+ parent,
+ properties={
+ 'children': relationship(
+ Child,
+ backref='parent')})
mapper(Child, child)
-
def test_attribute_set(self):
Parent, Child = self.classes.Parent, self.classes.Child
p1 = Parent()
for child in children:
p1.children.remove(child)
go()
-
-from sqlalchemy import *
-from sqlalchemy.testing import *
+from sqlalchemy.testing import fixtures, AssertsExecutionResults, profiling
from sqlalchemy.pool import QueuePool
from sqlalchemy import pool as pool_module
+pool = None
+
+
class QueuePoolTest(fixtures.TestBase, AssertsExecutionResults):
__requires__ = 'cpython',
class Connection(object):
+
def rollback(self):
pass
# class-level event listeners on Pool,
# if not present already.
p1 = QueuePool(creator=self.Connection,
- pool_size=3, max_overflow=-1,
- use_threadlocal=True)
+ pool_size=3, max_overflow=-1,
+ use_threadlocal=True)
p1.connect()
global pool
pool_size=3, max_overflow=-1,
use_threadlocal=True)
-
@profiling.function_call_count()
def test_first_connect(self):
- conn = pool.connect()
+ pool.connect()
def test_second_connect(self):
conn = pool.connect()
def go():
conn2 = pool.connect()
return conn2
- c2 = go()
+ go()
def test_second_samethread_connect(self):
conn = pool.connect()
+ conn # strong ref
@profiling.function_call_count()
def go():
return pool.connect()
- c2 = go()
-
-
+ go()
-from sqlalchemy import *
+from sqlalchemy import MetaData, Table, Column, String, Unicode, Integer, \
+ create_engine
from sqlalchemy.testing import fixtures, AssertsExecutionResults, profiling
from sqlalchemy import testing
from sqlalchemy.testing import eq_
NUM_FIELDS = 10
NUM_RECORDS = 1000
+t = t2 = metadata = None
+
class ResultSetTest(fixtures.TestBase, AssertsExecutionResults):
__backend__ = True
global t, t2, metadata
metadata = MetaData(testing.db)
t = Table('table', metadata, *[Column('field%d' % fnum, String(50))
- for fnum in range(NUM_FIELDS)])
- t2 = Table('table2', metadata, *[Column('field%d' % fnum,
- Unicode(50)) for fnum in range(NUM_FIELDS)])
+ for fnum in range(NUM_FIELDS)])
+ t2 = Table(
+ 'table2', metadata, *
+ [Column('field%d' % fnum, Unicode(50))
+ for fnum in range(NUM_FIELDS)])
def setup(self):
metadata.create_all()
t.insert().execute([dict(('field%d' % fnum, u('value%d' % fnum))
- for fnum in range(NUM_FIELDS)) for r_num in
- range(NUM_RECORDS)])
- t2.insert().execute([dict(('field%d' % fnum, u('value%d' % fnum))
- for fnum in range(NUM_FIELDS)) for r_num in
+ for fnum in range(NUM_FIELDS)) for r_num in
range(NUM_RECORDS)])
+ t2.insert().execute([dict(('field%d' % fnum, u('value%d' % fnum))
+ for fnum in range(NUM_FIELDS)) for r_num in
+ range(NUM_RECORDS)])
# warm up type caches
t.select().execute().fetchall()
def test_contains_doesnt_compile(self):
row = t.select().execute().first()
- c1 = Column('some column', Integer) + Column("some other column", Integer)
+ c1 = Column('some column', Integer) + \
+ Column("some other column", Integer)
+
@profiling.function_call_count()
def go():
c1 in row
def _rowproxy_fixture(self, keys, processors, row):
class MockMeta(object):
+
def __init__(self):
pass
keymap = {}
for index, (keyobjs, processor, values) in \
- enumerate(list(zip(keys, processors, row))):
+ enumerate(list(zip(keys, processors, row))):
for key in keyobjs:
keymap[key] = (processor, key, index)
keymap[index] = (processor, key, index)
def _test_getitem_value_refcounts(self, seq_factory):
col1, col2 = object(), object()
+
def proc1(value):
return value
value1, value2 = "x", "y"
def test_value_refcounts_custom_seq(self):
class CustomSeq(object):
+
def __init__(self, data):
self.data = data
import datetime
-import sys
-import time
-from sqlalchemy import *
+from sqlalchemy import Table, Column, Integer, Unicode, Date, \
+ DateTime, Time, Float, MetaData, Sequence, ForeignKey, create_engine, \
+ select, join, and_, outerjoin, func
from sqlalchemy.testing import fixtures, engines, profiling
from sqlalchemy import testing
ITERATIONS = 1
def test_baseline_0_setup(self):
global metadata
creator = testing.db.pool._creator
- recorder = lambda : dbapi_session.recorder(creator())
+ recorder = lambda: dbapi_session.recorder(creator())
engine = engines.testing_engine(options={'creator': recorder,
- 'use_reaper':False})
+ 'use_reaper': False})
metadata = MetaData(engine)
engine.connect()
def test_baseline_1_create_tables(self):
- Zoo = Table(
+ Table(
'Zoo',
metadata,
Column('ID', Integer, Sequence('zoo_id_seq'),
Column('Opens', Time),
Column('LastEscape', DateTime),
Column('Admission', Float),
- )
- Animal = Table(
+ )
+ Table(
'Animal',
metadata,
Column('ID', Integer, Sequence('animal_id_seq'),
Column('MotherID', Integer, ForeignKey('Animal.ID')),
Column('PreferredFoodID', Integer),
Column('AlternateFoodID', Integer),
- )
+ )
metadata.create_all()
def test_baseline_1a_populate(self):
Animal = metadata.tables['Animal']
engine = metadata.bind
wap = engine.execute(Zoo.insert(), Name='Wild Animal Park',
- Founded=datetime.date(2000, 1, 1),
- Opens=datetime.time(8, 15, 59),
- LastEscape=
- datetime.datetime(2004, 7, 29, 5, 6, 7),
- Admission=4.95).inserted_primary_key[0]
+ Founded=datetime.date(2000, 1, 1),
+ Opens=datetime.time(8, 15, 59),
+ LastEscape=datetime.datetime(
+ 2004, 7, 29, 5, 6, 7),
+ Admission=4.95).inserted_primary_key[0]
sdz = engine.execute(Zoo.insert(), Name='San Diego Zoo',
- Founded=datetime.date(1935, 9, 13),
- Opens=datetime.time(9, 0, 0),
- Admission=0).inserted_primary_key[0]
+ Founded=datetime.date(1935, 9, 13),
+ Opens=datetime.time(9, 0, 0),
+ Admission=0).inserted_primary_key[0]
engine.execute(Zoo.insert(inline=True), Name='Montr\xe9al Biod\xf4me',
- Founded=datetime.date(1992, 6, 19),
- Opens=datetime.time(9, 0, 0), Admission=11.75)
+ Founded=datetime.date(1992, 6, 19),
+ Opens=datetime.time(9, 0, 0), Admission=11.75)
seaworld = engine.execute(Zoo.insert(), Name='Sea_World',
- Admission=60).inserted_primary_key[0]
+ Admission=60).inserted_primary_key[0]
# Let's add a crazy futuristic Zoo to test large date values.
- lp = engine.execute(Zoo.insert(), Name='Luna Park',
- Founded=datetime.date(2072, 7, 17),
- Opens=datetime.time(0, 0, 0),
- Admission=134.95).inserted_primary_key[0]
+ engine.execute(
+ Zoo.insert(), Name='Luna Park',
+ Founded=datetime.date(2072, 7, 17),
+ Opens=datetime.time(0, 0, 0),
+ Admission=134.95).inserted_primary_key[0]
# Animals
leopardid = engine.execute(Animal.insert(), Species='Leopard',
- Lifespan=73.5).inserted_primary_key[0]
+ Lifespan=73.5).inserted_primary_key[0]
engine.execute(Animal.update(Animal.c.ID == leopardid), ZooID=wap,
- LastEscape=datetime.datetime( 2004, 12, 21, 8, 15, 0, 999907,)
- )
- lion = engine.execute(Animal.insert(), Species='Lion',
- ZooID=wap).inserted_primary_key[0]
+ LastEscape=datetime.datetime(
+ 2004, 12, 21, 8, 15, 0, 999907,)
+ )
+ engine.execute(
+ Animal.insert(),
+ Species='Lion', ZooID=wap).inserted_primary_key[0]
+
engine.execute(Animal.insert(), Species='Slug', Legs=1, Lifespan=.75)
- tiger = engine.execute(Animal.insert(), Species='Tiger',
- ZooID=sdz).inserted_primary_key[0]
+ engine.execute(Animal.insert(), Species='Tiger',
+ ZooID=sdz).inserted_primary_key[0]
# Override Legs.default with itself just to make sure it works.
engine.execute(Animal.insert(inline=True), Species='Bear', Legs=4)
engine.execute(Animal.insert(inline=True), Species='Ostrich', Legs=2,
- Lifespan=103.2)
+ Lifespan=103.2)
engine.execute(Animal.insert(inline=True), Species='Centipede',
- Legs=100)
- emp = engine.execute(Animal.insert(), Species='Emperor Penguin',
- Legs=2, ZooID=seaworld).inserted_primary_key[0]
- adelie = engine.execute(Animal.insert(), Species='Adelie Penguin',
- Legs=2, ZooID=seaworld).inserted_primary_key[0]
+ Legs=100)
+ engine.execute(Animal.insert(), Species='Emperor Penguin',
+ Legs=2, ZooID=seaworld).inserted_primary_key[0]
+ engine.execute(Animal.insert(), Species='Adelie Penguin',
+ Legs=2, ZooID=seaworld).inserted_primary_key[0]
engine.execute(Animal.insert(inline=True), Species='Millipede',
- Legs=1000000, ZooID=sdz)
+ Legs=1000000, ZooID=sdz)
# Add a mother and child to test relationships
- bai_yun = engine.execute(Animal.insert(), Species='Ape',
- Name='Bai Yun', Legs=2).inserted_primary_key[0]
+ bai_yun = engine.execute(
+ Animal.insert(),
+ Species='Ape',
+ Name='Bai Yun',
+ Legs=2).inserted_primary_key[0]
engine.execute(Animal.insert(inline=True), Species='Ape',
- Name='Hua Mei', Legs=2, MotherID=bai_yun)
+ Name='Hua Mei', Legs=2, MotherID=bai_yun)
def test_baseline_2_insert(self):
Animal = metadata.tables['Animal']
i = Animal.insert(inline=True)
for x in range(ITERATIONS):
- tick = i.execute(Species='Tick', Name='Tick %d' % x,
- Legs=8)
+ i.execute(Species='Tick', Name='Tick %d' % x, Legs=8)
def test_baseline_3_properties(self):
Zoo = metadata.tables['Zoo']
# Zoos
- WAP = fullobject(Zoo.select(Zoo.c.Name
- == 'Wild Animal Park'))
- SDZ = fullobject(Zoo.select(Zoo.c.Founded
- == datetime.date(1935, 9, 13)))
- Biodome = fullobject(Zoo.select(Zoo.c.Name
- == 'Montr\xe9al Biod\xf4me'))
- seaworld = fullobject(Zoo.select(Zoo.c.Admission
- == float(60)))
+ fullobject(Zoo.select(Zoo.c.Name == 'Wild Animal Park'))
+ fullobject(Zoo.select(Zoo.c.Founded ==
+ datetime.date(1935, 9, 13)))
+ fullobject(Zoo.select(Zoo.c.Name ==
+ 'Montr\xe9al Biod\xf4me'))
+ fullobject(Zoo.select(Zoo.c.Admission == float(60)))
# Animals
- leopard = fullobject(Animal.select(Animal.c.Species
- == 'Leopard'))
- ostrich = fullobject(Animal.select(Animal.c.Species
- == 'Ostrich'))
- millipede = fullobject(Animal.select(Animal.c.Legs
- == 1000000))
- ticks = fullobject(Animal.select(Animal.c.Species == 'Tick'
- ))
+ fullobject(Animal.select(Animal.c.Species == 'Leopard'))
+ fullobject(Animal.select(Animal.c.Species == 'Ostrich'))
+ fullobject(Animal.select(Animal.c.Legs == 1000000))
+ fullobject(Animal.select(Animal.c.Species == 'Tick'))
def test_baseline_4_expressions(self):
Zoo = metadata.tables['Zoo']
== 4
assert len(fulltable(Animal.select(Animal.c.Legs == 2))) \
== 5
- assert len(fulltable(Animal.select(and_(Animal.c.Legs >= 2,
- Animal.c.Legs < 20)))) == ITERATIONS + 9
+ assert len(
+ fulltable(
+ Animal.select(
+ and_(
+ Animal.c.Legs >= 2,
+ Animal.c.Legs < 20)))) == ITERATIONS + 9
assert len(fulltable(Animal.select(Animal.c.Legs > 10))) \
== 2
assert len(fulltable(Animal.select(Animal.c.Lifespan
- > 70))) == 2
+ > 70))) == 2
assert len(fulltable(Animal.select(Animal.c.Species.
- startswith('L')))) == 2
+ startswith('L')))) == 2
assert len(fulltable(Animal.select(Animal.c.Species.
- endswith('pede')))) == 2
+ endswith('pede')))) == 2
assert len(fulltable(Animal.select(Animal.c.LastEscape
- != None))) == 1
- assert len(fulltable(Animal.select(None
- == Animal.c.LastEscape))) == ITERATIONS + 11
+ != None))) == 1
+ assert len(
+ fulltable(
+ Animal.select(
+ None == Animal.c.LastEscape))) == ITERATIONS + 11
# In operator (containedby)
assert len(fulltable(Animal.select(Animal.c.Species.like('%pede%'
- )))) == 2
- assert len(fulltable(Animal.select(Animal.c.Species.in_(['Lion'
- , 'Tiger', 'Bear'])))) == 3
+ )))) == 2
+ assert len(
+ fulltable(
+ Animal.select(
+ Animal.c.Species.in_(
+ ['Lion', 'Tiger', 'Bear'])))) == 3
# Try In with cell references
class thing(object):
pass
-
pet, pet2 = thing(), thing()
pet.Name, pet2.Name = 'Slug', 'Ostrich'
- assert len(fulltable(Animal.select(Animal.c.Species.in_([pet.Name,
- pet2.Name])))) == 2
+ assert len(
+ fulltable(
+ Animal.select(
+ Animal.c.Species.in_([pet.Name, pet2.Name])))) == 2
# logic and other functions
assert len(fulltable(Animal.select(Animal.c.Species.like('Slug'
- )))) == 1
+ )))) == 1
assert len(fulltable(Animal.select(Animal.c.Species.like('%pede%'
- )))) == 2
+ )))) == 2
name = 'Lion'
- assert len(fulltable(Animal.select(func.length(Animal.c.Species)
- == len(name)))) == ITERATIONS + 3
- assert len(fulltable(Animal.select(Animal.c.Species.like('%i%'
- )))) == ITERATIONS + 7
+ assert len(
+ fulltable(
+ Animal.select(
+ func.length(
+ Animal.c.Species) == len(name)))) == ITERATIONS + 3
+ assert len(
+ fulltable(
+ Animal.select(
+ Animal.c.Species.like('%i%')))) == ITERATIONS + 7
# Test now(), today(), year(), month(), day()
- assert len(fulltable(Zoo.select(and_(Zoo.c.Founded != None,
- Zoo.c.Founded
- < func.current_timestamp(_type=Date))))) == 3
- assert len(fulltable(Animal.select(Animal.c.LastEscape
- == func.current_timestamp(_type=Date)))) == 0
- assert len(fulltable(Animal.select(func.date_part('year',
- Animal.c.LastEscape) == 2004))) == 1
- assert len(fulltable(Animal.select(func.date_part('month',
- Animal.c.LastEscape) == 12))) == 1
- assert len(fulltable(Animal.select(func.date_part('day',
- Animal.c.LastEscape) == 21))) == 1
+ assert len(
+ fulltable(
+ Zoo.select(
+ and_(
+ Zoo.c.Founded != None,
+ Zoo.c.Founded < func.current_timestamp(
+ _type=Date))))) == 3
+ assert len(
+ fulltable(
+ Animal.select(
+ Animal.c.LastEscape == func.current_timestamp(
+ _type=Date)))) == 0
+ assert len(
+ fulltable(
+ Animal.select(
+ func.date_part(
+ 'year',
+ Animal.c.LastEscape) == 2004))) == 1
+ assert len(
+ fulltable(
+ Animal.select(
+ func.date_part(
+ 'month',
+ Animal.c.LastEscape) == 12))) == 1
+ assert len(
+ fulltable(
+ Animal.select(
+ func.date_part(
+ 'day',
+ Animal.c.LastEscape) == 21))) == 1
def test_baseline_5_aggregates(self):
Animal = metadata.tables['Animal']
# views
view = engine.execute(select([Animal.c.Legs])).fetchall()
- legs = [x[0] for x in view]
- legs.sort()
+ legs = sorted([x[0] for x in view])
expected = {
'Leopard': 73.5,
'Slug': .75,
'Millipede': None,
'Ape': None,
'Tick': None,
- }
- for species, lifespan in engine.execute(select([Animal.c.Species,
- Animal.c.Lifespan])).fetchall():
+ }
+ for species, lifespan in engine.execute(
+ select([Animal.c.Species, Animal.c.Lifespan])).fetchall():
assert lifespan == expected[species]
expected = ['Montr\xe9al Biod\xf4me', 'Wild Animal Park']
- e = select([Zoo.c.Name], and_(Zoo.c.Founded != None,
- Zoo.c.Founded <= func.current_timestamp(),
- Zoo.c.Founded >= datetime.date(1990, 1, 1)))
+ e = select([Zoo.c.Name],
+ and_(Zoo.c.Founded != None,
+ Zoo.c.Founded <= func.current_timestamp(),
+ Zoo.c.Founded >= datetime.date(1990,
+ 1,
+ 1)))
values = [val[0] for val in engine.execute(e).fetchall()]
assert set(values) == set(expected)
# distinct
- legs = [x[0] for x in engine.execute(select([Animal.c.Legs],
- distinct=True)).fetchall()]
+ legs = [
+ x[0]
+ for x in engine.execute(
+ select([Animal.c.Legs],
+ distinct=True)).fetchall()]
legs.sort()
def test_baseline_6_editing(self):
# Edit
SDZ = engine.execute(Zoo.select(Zoo.c.Name == 'San Diego Zoo'
- )).first()
- engine.execute(Zoo.update(Zoo.c.ID == SDZ['ID'
- ]), Name='The San Diego Zoo',
- Founded=datetime.date(1900, 1, 1),
- Opens=datetime.time(7, 30, 0),
- Admission='35.00')
+ )).first()
+ engine.execute(
+ Zoo.update(
+ Zoo.c.ID == SDZ['ID']),
+ Name='The San Diego Zoo',
+ Founded=datetime.date(1900, 1, 1),
+ Opens=datetime.time(7, 30, 0), Admission='35.00')
# Test edits
SDZ = engine.execute(Zoo.select(Zoo.c.Name == 'The San Diego Zoo'
- )).first()
+ )).first()
assert SDZ['Founded'] == datetime.date(1900, 1, 1), \
SDZ['Founded']
# Change it back
engine.execute(Zoo.update(Zoo.c.ID == SDZ['ID'
- ]), Name='San Diego Zoo',
- Founded=datetime.date(1935, 9, 13),
- Opens=datetime.time(9, 0, 0),
- Admission='0')
+ ]), Name='San Diego Zoo',
+ Founded=datetime.date(1935, 9, 13),
+ Opens=datetime.time(9, 0, 0),
+ Admission='0')
# Test re-edits
SDZ = engine.execute(Zoo.select(Zoo.c.Name == 'San Diego Zoo'
- )).first()
+ )).first()
assert SDZ['Founded'] == datetime.date(1935, 9, 13)
def test_baseline_7_multiview(self):
return [list(row) for row in engine.execute(select).fetchall()]
for x in range(ITERATIONS):
- za = fulltable(select([Zoo.c.ID] + list(Animal.c),
- Zoo.c.Name == 'San Diego Zoo',
- from_obj=[join(Zoo, Animal)]))
- SDZ = Zoo.select(Zoo.c.Name == 'San Diego Zoo')
- e = fulltable(select([Zoo.c.ID, Animal.c.ID],
- and_(Zoo.c.Name == 'San Diego Zoo',
- Animal.c.Species == 'Leopard'),
- from_obj=[join(Zoo, Animal)]))
+ fulltable(
+ select(
+ [Zoo.c.ID] + list(Animal.c),
+ Zoo.c.Name == 'San Diego Zoo',
+ from_obj=[join(Zoo, Animal)]))
+ Zoo.select(Zoo.c.Name == 'San Diego Zoo')
+ fulltable(
+ select(
+ [Zoo.c.ID, Animal.c.ID],
+ and_(
+ Zoo.c.Name == 'San Diego Zoo',
+ Animal.c.Species == 'Leopard'
+ ),
+ from_obj=[join(Zoo, Animal)])
+ )
# Now try the same query with INNER, LEFT, and RIGHT JOINs.
- e = fulltable(select([Zoo.c.Name, Animal.c.Species],
- from_obj=[join(Zoo, Animal)]))
- e = fulltable(select([Zoo.c.Name, Animal.c.Species],
- from_obj=[outerjoin(Zoo, Animal)]))
- e = fulltable(select([Zoo.c.Name, Animal.c.Species],
- from_obj=[outerjoin(Animal, Zoo)]))
+ fulltable(select([
+ Zoo.c.Name, Animal.c.Species],
+ from_obj=[join(Zoo, Animal)]))
+ fulltable(select([
+ Zoo.c.Name, Animal.c.Species],
+ from_obj=[outerjoin(Zoo, Animal)]))
+ fulltable(select([
+ Zoo.c.Name, Animal.c.Species],
+ from_obj=[outerjoin(Animal, Zoo)]))
def test_baseline_8_drop(self):
metadata.drop_all()
def test_profile_0(self):
global metadata
- player = lambda : dbapi_session.player()
+ player = lambda: dbapi_session.player()
engine = create_engine('postgresql:///', creator=player,
- use_native_hstore=False)
+ use_native_hstore=False)
metadata = MetaData(engine)
engine.connect()
import datetime
-import sys
-import time
-from sqlalchemy import *
-from sqlalchemy.orm import *
+from sqlalchemy import Table, Column, Integer, Unicode, Date, \
+ DateTime, Time, Float, MetaData, Sequence, ForeignKey, create_engine, \
+ select, and_, func
+from sqlalchemy.orm import sessionmaker, mapper
from sqlalchemy.testing import fixtures, engines, profiling
from sqlalchemy import testing
ITERATIONS = 1
dbapi_session = engines.ReplayableSession()
metadata = None
+Zoo = Animal = session = None
+
class ZooMarkTest(fixtures.TestBase):
def test_baseline_0_setup(self):
global metadata, session
creator = testing.db.pool._creator
- recorder = lambda : dbapi_session.recorder(creator())
- engine = engines.testing_engine(options={'creator': recorder, 'use_reaper':False})
+ recorder = lambda: dbapi_session.recorder(creator())
+ engine = engines.testing_engine(
+ options={'creator': recorder, 'use_reaper': False})
metadata = MetaData(engine)
session = sessionmaker(engine)()
engine.connect()
Column('Opens', Time),
Column('LastEscape', DateTime),
Column('Admission', Float),
- )
+ )
animal = Table(
'Animal',
metadata,
Column('MotherID', Integer, ForeignKey('Animal.ID')),
Column('PreferredFoodID', Integer),
Column('AlternateFoodID', Integer),
- )
+ )
metadata.create_all()
global Zoo, Animal
-
class Zoo(object):
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
-
class Animal(object):
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
-
mapper(Zoo, zoo)
mapper(Animal, animal)
def test_baseline_1a_populate(self):
- wap = Zoo(Name='Wild Animal Park', Founded=datetime.date(2000,
- 1, 1), Opens=datetime.time(8, 15, 59),
- LastEscape=datetime.datetime( 2004, 7, 29, 5, 6, 7, ),
- Admission=4.95)
+ wap = Zoo(
+ Name='Wild Animal Park', Founded=datetime.date(
+ 2000, 1, 1), Opens=datetime.time(
+ 8, 15, 59), LastEscape=datetime.datetime(
+ 2004, 7, 29, 5, 6, 7, ), Admission=4.95)
session.add(wap)
- sdz = Zoo(Name='San Diego Zoo', Founded=datetime.date(1835, 9,
- 13), Opens=datetime.time(9, 0, 0), Admission=0)
+ sdz = Zoo(
+ Name='San Diego Zoo', Founded=datetime.date(
+ 1835, 9, 13), Opens=datetime.time(
+ 9, 0, 0), Admission=0)
session.add(sdz)
bio = Zoo(Name='Montr\xe9al Biod\xf4me',
Founded=datetime.date(1992, 6, 19),
session.add(leopard)
leopard.ZooID = wap.ID
leopard.LastEscape = \
- datetime.datetime(2004, 12, 21, 8, 15, 0, 999907, )
+ datetime.datetime(2004, 12, 21, 8, 15, 0, 999907, )
session.add(Animal(Species='Lion', ZooID=wap.ID))
session.add(Animal(Species='Slug', Legs=1, Lifespan=.75))
session.add(Animal(Species='Tiger', ZooID=sdz.ID))
session.add(Animal(Species='Ostrich', Legs=2, Lifespan=103.2))
session.add(Animal(Species='Centipede', Legs=100))
session.add(Animal(Species='Emperor Penguin', Legs=2,
- ZooID=seaworld.ID))
+ ZooID=seaworld.ID))
session.add(Animal(Species='Adelie Penguin', Legs=2,
- ZooID=seaworld.ID))
+ ZooID=seaworld.ID))
session.add(Animal(Species='Millipede', Legs=1000000,
- ZooID=sdz.ID))
+ ZooID=sdz.ID))
# Add a mother and child to test relationships
bai_yun = Animal(Species='Ape', Nameu='Bai Yun', Legs=2)
session.add(bai_yun)
session.add(Animal(Species='Ape', Name='Hua Mei', Legs=2,
- MotherID=bai_yun.ID))
+ MotherID=bai_yun.ID))
session.flush()
session.commit()
def test_baseline_2_insert(self):
for x in range(ITERATIONS):
session.add(Animal(Species='Tick', Name='Tick %d' % x,
- Legs=8))
+ Legs=8))
session.flush()
def test_baseline_3_properties(self):
# Zoos
- WAP = list(session.query(Zoo).filter(Zoo.Name
- == 'Wild Animal Park'))
- SDZ = list(session.query(Zoo).filter(Zoo.Founded
- == datetime.date(1835, 9, 13)))
- Biodome = list(session.query(Zoo).filter(Zoo.Name
- == 'Montr\xe9al Biod\xf4me'))
- seaworld = list(session.query(Zoo).filter(Zoo.Admission
- == float(60)))
+ list(session.query(Zoo).filter(
+ Zoo.Name == 'Wild Animal Park'))
+ list(
+ session.query(Zoo).filter(
+ Zoo.Founded == datetime.date(
+ 1835,
+ 9,
+ 13)))
+ list(
+ session.query(Zoo).filter(
+ Zoo.Name == 'Montr\xe9al Biod\xf4me'))
+ list(session.query(Zoo).filter(Zoo.Admission == float(60)))
# Animals
- leopard = list(session.query(Animal).filter(Animal.Species
- == 'Leopard'))
- ostrich = list(session.query(Animal).filter(Animal.Species
- == 'Ostrich'))
- millipede = list(session.query(Animal).filter(Animal.Legs
- == 1000000))
- ticks = list(session.query(Animal).filter(Animal.Species
- == 'Tick'))
+ list(session.query(Animal).filter(Animal.Species == 'Leopard'))
+ list(session.query(Animal).filter(Animal.Species == 'Ostrich'))
+ list(session.query(Animal).filter(Animal.Legs == 1000000))
+ list(session.query(Animal).filter(Animal.Species == 'Tick'))
def test_baseline_4_expressions(self):
for x in range(ITERATIONS):
assert len(list(session.query(Zoo))) == 5
assert len(list(session.query(Animal))) == ITERATIONS + 12
assert len(list(session.query(Animal).filter(Animal.Legs
- == 4))) == 4
+ == 4))) == 4
assert len(list(session.query(Animal).filter(Animal.Legs
- == 2))) == 5
- assert len(list(session.query(Animal).filter(and_(Animal.Legs
- >= 2, Animal.Legs < 20)))) == ITERATIONS + 9
+ == 2))) == 5
+ assert len(
+ list(
+ session.query(Animal).filter(
+ and_(
+ Animal.Legs >= 2,
+ Animal.Legs < 20)))) == ITERATIONS + 9
assert len(list(session.query(Animal).filter(Animal.Legs
- > 10))) == 2
+ > 10))) == 2
assert len(list(session.query(Animal).filter(Animal.Lifespan
- > 70))) == 2
+ > 70))) == 2
assert len(list(session.query(Animal).
- filter(Animal.Species.like('L%')))) == 2
+ filter(Animal.Species.like('L%')))) == 2
assert len(list(session.query(Animal).
- filter(Animal.Species.like('%pede')))) == 2
- assert len(list(session.query(Animal).filter(Animal.LastEscape
- != None))) == 1
+ filter(Animal.Species.like('%pede')))) == 2
assert len(list(session.query(Animal).filter(Animal.LastEscape
- == None))) == ITERATIONS + 11
+ != None))) == 1
+ assert len(
+ list(
+ session.query(Animal).filter(
+ Animal.LastEscape == None))) == ITERATIONS + 11
# In operator (containedby)
assert len(list(session.query(Animal).filter(
- Animal.Species.like('%pede%')))) == 2
- assert len(list(session.query(Animal).
- filter(Animal.Species.in_(('Lion'
- , 'Tiger', 'Bear'))))) == 3
+ Animal.Species.like('%pede%')))) == 2
+ assert len(
+ list(
+ session.query(Animal). filter(
+ Animal.Species.in_(
+ ('Lion', 'Tiger', 'Bear'))))) == 3
# Try In with cell references
class thing(object):
pet, pet2 = thing(), thing()
pet.Name, pet2.Name = 'Slug', 'Ostrich'
assert len(list(session.query(Animal).
- filter(Animal.Species.in_((pet.Name,
- pet2.Name))))) == 2
+ filter(Animal.Species.in_((pet.Name,
+ pet2.Name))))) == 2
# logic and other functions
name = 'Lion'
assert len(list(session.query(Animal).
- filter(func.length(Animal.Species)
- == len(name)))) == ITERATIONS + 3
+ filter(func.length(Animal.Species)
+ == len(name)))) == ITERATIONS + 3
assert len(list(session.query(Animal).
- filter(Animal.Species.like('%i%'
- )))) == ITERATIONS + 7
+ filter(Animal.Species.like('%i%'
+ )))) == ITERATIONS + 7
# Test now(), today(), year(), month(), day()
- assert len(list(session.query(Zoo).filter(and_(Zoo.Founded
- != None, Zoo.Founded < func.now())))) == 3
+ assert len(
+ list(
+ session.query(Zoo).filter(
+ and_(
+ Zoo.Founded != None,
+ Zoo.Founded < func.now())))) == 3
assert len(list(session.query(Animal).filter(Animal.LastEscape
- == func.now()))) == 0
- assert len(list(session.query(Animal).filter(func.date_part('year'
- , Animal.LastEscape) == 2004))) == 1
- assert len(list(session.query(Animal).
- filter(func.date_part('month'
- , Animal.LastEscape) == 12))) == 1
- assert len(list(session.query(Animal).filter(func.date_part('day'
- , Animal.LastEscape) == 21))) == 1
+ == func.now()))) == 0
+ assert len(list(session.query(Animal).filter(
+ func.date_part('year', Animal.LastEscape) == 2004))) == 1
+ assert len(
+ list(
+ session.query(Animal). filter(
+ func.date_part(
+ 'month',
+ Animal.LastEscape) == 12))) == 1
+ assert len(list(session.query(Animal).filter(
+ func.date_part('day', Animal.LastEscape) == 21))) == 1
def test_baseline_5_aggregates(self):
Animal = metadata.tables['Animal']
# views
view = engine.execute(select([Animal.c.Legs])).fetchall()
- legs = [x[0] for x in view]
- legs.sort()
+ legs = sorted([x[0] for x in view])
expected = {
'Leopard': 73.5,
'Slug': .75,
'Millipede': None,
'Ape': None,
'Tick': None,
- }
- for species, lifespan in engine.execute(select([Animal.c.Species,
- Animal.c.Lifespan])).fetchall():
+ }
+ for species, lifespan in engine.execute(
+ select([Animal.c.Species, Animal.c.Lifespan])).fetchall():
assert lifespan == expected[species]
expected = ['Montr\xe9al Biod\xf4me', 'Wild Animal Park']
- e = select([Zoo.c.Name], and_(Zoo.c.Founded != None,
- Zoo.c.Founded <= func.current_timestamp(),
- Zoo.c.Founded >= datetime.date(1990, 1, 1)))
+ e = select([Zoo.c.Name],
+ and_(Zoo.c.Founded != None,
+ Zoo.c.Founded <= func.current_timestamp(),
+ Zoo.c.Founded >= datetime.date(1990,
+ 1,
+ 1)))
values = [val[0] for val in engine.execute(e).fetchall()]
assert set(values) == set(expected)
# distinct
- legs = [x[0] for x in engine.execute(select([Animal.c.Legs],
- distinct=True)).fetchall()]
+ legs = [
+ x[0]
+ for x in engine.execute(
+ select([Animal.c.Legs],
+ distinct=True)).fetchall()]
legs.sort()
def test_baseline_6_editing(self):
# Edit
SDZ = session.query(Zoo).filter(Zoo.Name == 'San Diego Zoo'
- ).one()
+ ).one()
SDZ.Name = 'The San Diego Zoo'
SDZ.Founded = datetime.date(1900, 1, 1)
SDZ.Opens = datetime.time(7, 30, 0)
# Test edits
SDZ = session.query(Zoo).filter(Zoo.Name
- == 'The San Diego Zoo').one()
+ == 'The San Diego Zoo').one()
assert SDZ.Founded == datetime.date(1900, 1, 1), SDZ.Founded
# Change it back
# Test re-edits
SDZ = session.query(Zoo).filter(Zoo.Name == 'San Diego Zoo'
- ).one()
+ ).one()
assert SDZ.Founded == datetime.date(1835, 9, 13), \
SDZ.Founded
def test_profile_0(self):
global metadata, session
- player = lambda : dbapi_session.player()
+ player = lambda: dbapi_session.player()
engine = create_engine('postgresql:///', creator=player,
- use_native_hstore=False)
+ use_native_hstore=False)
metadata = MetaData(engine)
session = sessionmaker(engine)()
engine.connect()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
