clear_mappers()
objectstore.clear()
global Person, Preferences, Address
-
+
class Person(ActiveMapper):
class mapping:
__version_id_col__ = 'row_version'
class Address(ActiveMapper):
class mapping:
- # note that in other objects, the 'id' primary key is
+ # note that in other objects, the 'id' primary key is
# automatically added -- if you specify a primary key,
# then ActiveMapper will not add an integer primary key
# for you.
postal_code = column(String(128))
person_id = column(Integer, foreign_key=ForeignKey('person.id'))
- activemapper.metadata.connect(testbase.db)
+ activemapper.metadata.bind = testbase.db
activemapper.create_tables()
-
+
def tearDownAll(self):
clear_mappers()
activemapper.drop_tables()
-
+
def tearDown(self):
for t in activemapper.metadata.table_iterator(reverse=True):
t.delete().execute()
-
+
def create_person_one(self):
# create a person
p1 = Person(
]
)
return p1
-
-
+
+
def create_person_two(self):
p2 = Person(
full_name='Lacey LaCour',
# a "self.preferences = Preferences()" into the __init__
# of Person also doens't seem to fix this
p2.preferences = Preferences()
-
+
return p2
-
-
+
+
def test_create(self):
p1 = self.create_person_one()
objectstore.flush()
objectstore.clear()
-
- results = Person.query.select()
-
+
+ results = Person.query.all()
+
self.assertEquals(len(results), 1)
-
+
person = results[0]
self.assertEquals(person.id, p1.id)
self.assertEquals(len(person.addresses), 2)
p1 = self.create_person_one()
objectstore.flush()
objectstore.clear()
-
- person = Person.query.select()[0]
+
+ person = Person.query.first()
person.gender = 'F'
objectstore.flush()
objectstore.clear()
self.assertEquals(person.row_version, 2)
- person = Person.query.select()[0]
+ person = Person.query.first()
person.gender = 'M'
objectstore.flush()
objectstore.clear()
self.assertEquals(person.row_version, 3)
#TODO: check that a concurrent modification raises exception
- p1 = Person.query.select()[0]
+ p1 = Person.query.first()
s1 = objectstore()
s2 = create_session()
objectstore.registry.set(s2)
- p2 = Person.query.select()[0]
+ p2 = Person.query.first()
p1.first_name = "jack"
p2.first_name = "ed"
objectstore.flush()
assert False
except exceptions.ConcurrentModificationError:
pass
-
-
+
+
def test_delete(self):
p1 = self.create_person_one()
-
+
objectstore.flush()
objectstore.clear()
-
- results = Person.query.select()
+
+ results = Person.query.all()
self.assertEquals(len(results), 1)
-
+
objectstore.delete(results[0])
objectstore.flush()
objectstore.clear()
-
- results = Person.query.select()
+
+ results = Person.query.all()
self.assertEquals(len(results), 0)
-
-
+
+
def test_multiple(self):
p1 = self.create_person_one()
p2 = self.create_person_two()
-
+
objectstore.flush()
objectstore.clear()
-
+
# select and make sure we get back two results
- people = Person.query.select()
+ people = Person.query.all()
self.assertEquals(len(people), 2)
-
+
# make sure that our backwards relationships work
self.assertEquals(people[0].addresses[0].person.id, p1.id)
self.assertEquals(people[1].addresses[0].person.id, p2.id)
-
+
# try a more complex select
- results = Person.query.select(
+ results = Person.query.filter(
or_(
and_(
Address.c.person_id == Person.c.id,
Preferences.c.favorite_color == 'Green'
)
)
- )
+ ).all()
self.assertEquals(len(results), 2)
-
-
+
+
def test_oneway_backref(self):
# FIXME: I don't know why, but it seems that my backwards relationship
# on preferences still ends up being a list even though I pass
self.assertEquals(p1.preferences.person, p1)
objectstore.flush()
objectstore.delete(p1)
-
+
objectstore.flush()
objectstore.clear()
-
-
+
+
def test_select_by(self):
# FIXME: either I don't understand select_by, or it doesn't work.
# FIXED (as good as we can for now): yup....everyone thinks it works that way....it only
# generates joins for keyword arguments, not ColumnClause args. would need a new layer of
# "MapperClause" objects to use properties in expressions. (MB)
-
+
p1 = self.create_person_one()
p2 = self.create_person_two()
-
+
objectstore.flush()
objectstore.clear()
-
- results = Person.query.join('addresses').select(
- Address.c.postal_code.like('30075')
- )
+
+ results = Person.query.join('addresses').filter(
+ Address.c.postal_code.like('30075')
+ ).all()
self.assertEquals(len(results), 1)
self.assertEquals(Person.query.count(), 2)
name = column(String(30))
foorel = many_to_many("foo", secondarytable, backref='bazrel')
- activemapper.metadata.connect(testbase.db)
+ activemapper.metadata.bind = testbase.db
activemapper.create_tables()
# Create a couple of activemapper objects
objectstore.flush()
objectstore.clear()
- foo1 = foo.query.get_by(name='foo1')
- baz1 = baz.query.get_by(name='baz1')
-
+ foo1 = foo.query.filter_by(name='foo1').one()
+ baz1 = baz.query.filter_by(name='baz1').one()
+
# Just checking ...
assert (foo1.name == 'foo1')
assert (baz1.name == 'baz1')
# baz1 to foo1.bazrel - (AttributeError: 'foo' object has no attribute 'bazrel')
foo1.bazrel.append(baz1)
assert (foo1.bazrel == [baz1])
-
+
class testselfreferential(PersistTest):
def setUpAll(self):
clear_mappers()
name = column(String(30))
parent_id = column(Integer, foreign_key=ForeignKey('treenode.id'))
children = one_to_many('TreeNode', colname='id', backref='parent')
-
- activemapper.metadata.connect(testbase.db)
+
+ activemapper.metadata.bind = testbase.db
activemapper.create_tables()
def tearDownAll(self):
clear_mappers()
t.children.append(TreeNode(name='node3'))
objectstore.flush()
objectstore.clear()
-
- t = TreeNode.query.get_by(name='node1')
+
+ t = TreeNode.query.filter_by(name='node1').one()
assert (t.name == 'node1')
assert (t.children[0].name == 'node2')
assert (t.children[1].name == 'node3')
assert (t.children[1].parent is t)
objectstore.clear()
- t = TreeNode.query.get_by(name='node3')
- assert (t.parent is TreeNode.query.get_by(name='node1'))
-
+ t = TreeNode.query.filter_by(name='node3').one()
+ assert (t.parent is TreeNode.query.filter_by(name='node1').one())
+
if __name__ == '__main__':
testbase.main()
import testbase
-
+import warnings
from sqlalchemy import *
from sqlalchemy import exceptions
from sqlalchemy.orm import create_session, clear_mappers, relation, class_mapper
-from sqlalchemy.ext.assignmapper import assign_mapper
+import sqlalchemy.ext.assignmapper
from sqlalchemy.ext.sessioncontext import SessionContext
from testlib import *
+def assign_mapper(*args, **kw):
+ try:
+ warnings.filterwarnings('ignore', 'assign_mapper is deprecated')
+ sqlalchemy.ext.assignmapper.assign_mapper(*args, **kw)
+ finally:
+ warnings.filterwarnings('always', 'assign_mapper is deprecated')
class AssignMapperTest(PersistTest):
def setUpAll(self):
global metadata, table, table2
metadata = MetaData(testbase.db)
- table = Table('sometable', metadata,
+ table = Table('sometable', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)))
- table2 = Table('someothertable', metadata,
+ table2 = Table('someothertable', metadata,
Column('id', Integer, primary_key=True),
Column('someid', None, ForeignKey('sometable.id'))
)
global SomeObject, SomeOtherObject, ctx
class SomeObject(object):pass
class SomeOtherObject(object):pass
-
- ctx = SessionContext(create_session)
- assign_mapper(ctx, SomeObject, table, properties={
- 'options':relation(SomeOtherObject)
- })
- assign_mapper(ctx, SomeOtherObject, table2)
-
- s = SomeObject()
- s.id = 1
- s.data = 'hello'
- sso = SomeOtherObject()
- s.options.append(sso)
- ctx.current.flush()
- ctx.current.clear()
+
+ deps = ('SessionContext is deprecated',
+ 'SessionContextExt is deprecated')
+ try:
+ for dep in deps:
+ warnings.filterwarnings('ignore', dep)
+
+ ctx = SessionContext(create_session)
+ assign_mapper(ctx, SomeObject, table, properties={
+ 'options':relation(SomeOtherObject)
+ })
+ assign_mapper(ctx, SomeOtherObject, table2)
+
+ s = SomeObject()
+ s.id = 1
+ s.data = 'hello'
+ sso = SomeOtherObject()
+ s.options.append(sso)
+ ctx.current.flush()
+ ctx.current.clear()
+ finally:
+ for dep in deps:
+ warnings.filterwarnings('always', dep)
def tearDownAll(self):
metadata.drop_all()
for table in metadata.table_iterator(reverse=True):
table.delete().execute()
clear_mappers()
-
+
def test_override_attributes(self):
-
+
sso = SomeOtherObject.query().first()
-
+
assert SomeObject.query.filter_by(id=1).one().options[0].id == sso.id
s2 = SomeObject(someid=12)
s3 = SomeOtherObject(someid=123, bogus=345)
-
+
class ValidatedOtherObject(object):pass
assign_mapper(ctx, ValidatedOtherObject, table2, validate=True)
assert False
except exceptions.ArgumentError:
pass
-
+
def test_dont_clobber_methods(self):
class MyClass(object):
def expunge(self):
return "an expunge !"
-
+
assign_mapper(ctx, MyClass, table2)
-
+
assert MyClass().expunge() == "an expunge !"
-
+
if __name__ == '__main__':
testbase.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
