本文整理汇总了Python中sqlalchemy.test.testing.eq_函数的典型用法代码示例。如果您正苦于以下问题:Python eq_函数的具体用法?Python eq_怎么用?Python eq_使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了eq_函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_remove_orphans
def test_remove_orphans(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
cascade="all, delete-orphan", backref='user')
})
sess = create_session(autoflush=True)
u = User(name='ed')
u.addresses.append(Address(email_address='a'))
u.addresses.append(Address(email_address='b'))
u.addresses.append(Address(email_address='c'))
u.addresses.append(Address(email_address='d'))
u.addresses.append(Address(email_address='e'))
u.addresses.append(Address(email_address='f'))
sess.add(u)
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
Address(email_address='d'), Address(email_address='e'), Address(email_address='f')],
sess.query(Address).all())
eq_(Address(email_address='c'), u.addresses[2])
try:
del u.addresses[3]
assert False
except TypeError, e:
assert "doesn't support item deletion" in str(e), str(e)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:26,代码来源:test_dynamic.py
示例2: test_cascades_onlycollection
def test_cascades_onlycollection(self):
"""Cascade only reaches instances that are still part of the collection,
not those that have been removed"""
sess = create_session()
u = User(name='jack',
orders=[Order(description='someorder'),
Order(description='someotherorder')])
sess.add(u)
sess.flush()
o = u.orders[0]
del u.orders[0]
sess.delete(u)
assert u in sess.deleted
assert o not in sess.deleted
assert o in sess
u2 = User(name='newuser', orders=[o])
sess.add(u2)
sess.flush()
sess.expunge_all()
assert users.count().scalar() == 1
assert orders.count().scalar() == 1
eq_(sess.query(User).all(),
[User(name='newuser',
orders=[Order(description='someorder')])])
开发者ID:gajop,项目名称:springgrid,代码行数:27,代码来源:test_cascade.py
示例3: _test_decimal
def _test_decimal(self, tabledef):
"""Checks a variety of FIXED usages.
This is primarily for SERIAL columns, which can be FIXED (scale-less)
or (SMALL)INT. Ensures that FIXED id columns are converted to
integers and that are assignable as such. Also exercises general
decimal assignment and selection behavior.
"""
meta = MetaData(testing.db)
try:
if isinstance(tabledef, basestring):
# run textual CREATE TABLE
testing.db.execute(tabledef)
else:
_t = tabledef.tometadata(meta)
_t.create()
t = Table('dectest', meta, autoload=True)
vals = [Decimal('2.2'), Decimal('23'), Decimal('2.4'), 25]
cols = ['d1','d2','n1','i1']
t.insert().execute(dict(zip(cols,vals)))
roundtrip = list(t.select().execute())
eq_(roundtrip, [tuple([1] + vals)])
t.insert().execute(dict(zip(['id'] + cols,
[2] + list(roundtrip[0][1:]))))
roundtrip2 = list(t.select(order_by=t.c.id).execute())
eq_(roundtrip2, [tuple([1] + vals),
tuple([2] + vals)])
finally:
try:
testing.db.execute("DROP TABLE dectest")
except exc.DatabaseError:
pass
开发者ID:AndryulE,项目名称:kitsune,代码行数:35,代码来源:test_maxdb.py
示例4: test_nativeext_submanager
def test_nativeext_submanager(self):
class Mine(attributes.ClassManager): pass
class A(object):
__sa_instrumentation_manager__ = Mine
attributes.register_class(A)
eq_(type(attributes.manager_of_class(A)), Mine)
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:7,代码来源:test_instrumentation.py
示例5: test_parent_refs_descendant
def test_parent_refs_descendant(self):
class Person(AttrSettable):
pass
class Manager(Person):
pass
# note that up until recently (0.4.4), we had to specify "foreign_keys" here
# for this primary join.
mapper(Person, people, properties={
'manager':relation(Manager, primaryjoin=(people.c.manager_id ==
managers.c.person_id),
uselist=False, post_update=True)
})
mapper(Manager, managers, inherits=Person,
inherit_condition=people.c.person_id==managers.c.person_id)
eq_(class_mapper(Person).get_property('manager').synchronize_pairs, [(managers.c.person_id,people.c.manager_id)])
session = create_session()
p = Person(name='some person')
m = Manager(name='some manager')
p.manager = m
session.add(p)
session.flush()
session.expunge_all()
p = session.query(Person).get(p.person_id)
m = session.query(Manager).get(m.person_id)
print p, m, p.manager
assert p.manager is m
开发者ID:clones,项目名称:sqlalchemy,代码行数:30,代码来源:test_polymorph2.py
示例6: go
def go():
sess = create_session()
a1 = A(col2="a1")
a2 = A(col2="a2")
a3 = A(col2="a3")
a1.bs.append(B(col2="b1"))
a1.bs.append(B(col2="b2"))
a3.bs.append(B(col2="b3"))
for x in [a1,a2,a3]:
sess.add(x)
sess.flush()
sess.expunge_all()
alist = sess.query(A).all()
eq_(
[
A(col2="a1", bs=[B(col2="b1"), B(col2="b2")]),
A(col2="a2", bs=[]),
A(col2="a3", bs=[B(col2="b3")])
],
alist)
for a in alist:
sess.delete(a)
sess.flush()
开发者ID:clones,项目名称:sqlalchemy,代码行数:25,代码来源:test_memusage.py
示例7: test_clear
def test_clear(self):
uowcommit, a1, b1, a_mapper, b_mapper = self._fixture()
pairs = [(a_mapper.c.id, b_mapper.c.t1id,)]
b1.obj().t1id = 8
eq_(b1.obj().__dict__['t1id'], 8)
sync.clear(b1, b_mapper, pairs)
eq_(b1.obj().__dict__['t1id'], None)
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:7,代码来源:test_sync.py
示例8: test_nested_transaction_connection_add
def test_nested_transaction_connection_add(self):
mapper(User, users)
sess = create_session(autocommit=True)
sess.begin()
sess.begin_nested()
u1 = User(name='u1')
sess.add(u1)
sess.flush()
sess.rollback()
u2 = User(name='u2')
sess.add(u2)
sess.commit()
eq_(set(sess.query(User).all()), set([u2]))
sess.begin()
sess.begin_nested()
u3 = User(name='u3')
sess.add(u3)
sess.commit() # commit the nested transaction
sess.rollback()
eq_(set(sess.query(User).all()), set([u2]))
sess.close()
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:32,代码来源:test_session.py
示例9: test_mixed_transaction_control
def test_mixed_transaction_control(self):
mapper(User, users)
sess = create_session(autocommit=True)
sess.begin()
sess.begin_nested()
transaction = sess.begin(subtransactions=True)
sess.add(User(name='u1'))
transaction.commit()
sess.commit()
sess.commit()
sess.close()
eq_(len(sess.query(User).all()), 1)
t1 = sess.begin()
t2 = sess.begin_nested()
sess.add(User(name='u2'))
t2.commit()
assert sess.transaction is t1
sess.close()
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:28,代码来源:test_session.py
示例10: _test_onetoone
def _test_onetoone(self, passive_updates):
mapper(User, users, properties={
"address":relation(Address, passive_updates=passive_updates, uselist=False)
})
mapper(Address, addresses)
sess = create_session()
u1 = User(username='jack', fullname='jack')
sess.add(u1)
sess.flush()
a1 = Address(email='jack1')
u1.address = a1
sess.add(a1)
sess.flush()
u1.username = 'ed'
def go():
sess.flush()
if passive_updates:
sess.expire(u1, ['address'])
self.assert_sql_count(testing.db, go, 1)
else:
self.assert_sql_count(testing.db, go, 2)
def go():
sess.flush()
self.assert_sql_count(testing.db, go, 0)
sess.expunge_all()
eq_([Address(username='ed')], sess.query(Address).all())
开发者ID:gajop,项目名称:springgrid,代码行数:32,代码来源:test_naturalpks.py
示例11: test_before_flush_affects_dirty
def test_before_flush_affects_dirty(self):
mapper(User, users)
class MyExt(sa.orm.session.SessionExtension):
def before_flush(self, session, flush_context, objects):
for obj in list(session.identity_map.values()):
obj.name += " modified"
sess = create_session(extension = MyExt(), autoflush=True)
u = User(name='u1')
sess.add(u)
sess.flush()
eq_(sess.query(User).order_by(User.name).all(),
[
User(name='u1')
]
)
sess.add(User(name='u2'))
sess.flush()
sess.expunge_all()
eq_(sess.query(User).order_by(User.name).all(),
[
User(name='u1 modified'),
User(name='u2')
]
)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:27,代码来源:test_session.py
示例12: _test_manytoone
def _test_manytoone(self, passive_updates):
mapper(User, users)
mapper(Address, addresses, properties={
'user':relation(User, passive_updates=passive_updates)
})
sess = create_session()
a1 = Address(email='jack1')
a2 = Address(email='jack2')
u1 = User(username='jack', fullname='jack')
a1.user = u1
a2.user = u1
sess.add(a1)
sess.add(a2)
sess.flush()
u1.username = 'ed'
def go():
sess.flush()
if passive_updates:
self.assert_sql_count(testing.db, go, 1)
else:
self.assert_sql_count(testing.db, go, 3)
def go():
sess.flush()
self.assert_sql_count(testing.db, go, 0)
assert a1.username == a2.username == 'ed'
sess.expunge_all()
eq_([Address(username='ed'), Address(username='ed')], sess.query(Address).all())
开发者ID:gajop,项目名称:springgrid,代码行数:33,代码来源:test_naturalpks.py
示例13: test_reduce_aliased_join
def test_reduce_aliased_join(self):
metadata = MetaData()
people = Table('people', metadata,
Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
Column('name', String(50)),
Column('type', String(30)))
engineers = Table('engineers', metadata,
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('status', String(30)),
Column('engineer_name', String(50)),
Column('primary_language', String(50)),
)
managers = Table('managers', metadata,
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('status', String(30)),
Column('manager_name', String(50))
)
pjoin = people.outerjoin(engineers).outerjoin(managers).select(use_labels=True).alias('pjoin')
eq_(
util.column_set(sql_util.reduce_columns([pjoin.c.people_person_id, pjoin.c.engineers_person_id, pjoin.c.managers_person_id])),
util.column_set([pjoin.c.people_person_id])
)
开发者ID:clones,项目名称:sqlalchemy,代码行数:25,代码来源:test_selectable.py
示例14: test_count
def test_count(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u = sess.query(User).first()
eq_(u.addresses.count(), 1)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:7,代码来源:test_dynamic.py
示例15: test_of_type
def test_of_type(self):
mapper(Company, companies, properties={
'employees':relationship(Employee, backref='company')
})
mapper(Employee, employees, polymorphic_on=employees.c.type)
mapper(Manager, inherits=Employee, polymorphic_identity='manager')
mapper(Engineer, inherits=Employee, polymorphic_identity='engineer')
mapper(JuniorEngineer, inherits=Engineer, polymorphic_identity='juniorengineer')
sess = sessionmaker()()
c1 = Company(name='c1')
c2 = Company(name='c2')
m1 = Manager(name='Tom', manager_data='data1', company=c1)
m2 = Manager(name='Tom2', manager_data='data2', company=c2)
e1 = Engineer(name='Kurt', engineer_info='knows how to hack', company=c2)
e2 = JuniorEngineer(name='Ed', engineer_info='oh that ed', company=c1)
sess.add_all([c1, c2, m1, m2, e1, e2])
sess.commit()
sess.expunge_all()
eq_(
sess.query(Company).filter(Company.employees.of_type(JuniorEngineer).any()).all(),
[
Company(name='c1'),
]
)
eq_(
sess.query(Company).join(Company.employees.of_type(JuniorEngineer)).all(),
[
Company(name='c1'),
]
)
开发者ID:AndryulE,项目名称:kitsune,代码行数:33,代码来源:test_single.py
示例16: test_weakref_with_cycles_o2o
def test_weakref_with_cycles_o2o(self):
s = sessionmaker()()
mapper(User, users, properties={
"address":relationship(Address, backref="user", uselist=False)
})
mapper(Address, addresses)
s.add(User(name="ed", address=Address(email_address="ed1")))
s.commit()
user = s.query(User).options(joinedload(User.address)).one()
user.address.user
eq_(user, User(name="ed", address=Address(email_address="ed1")))
del user
gc_collect()
assert len(s.identity_map) == 0
user = s.query(User).options(joinedload(User.address)).one()
user.address.email_address='ed2'
user.address.user # lazyload
del user
gc_collect()
assert len(s.identity_map) == 2
s.commit()
user = s.query(User).options(joinedload(User.address)).one()
eq_(user, User(name="ed", address=Address(email_address="ed2")))
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:28,代码来源:test_session.py
示例17: test_non_orphan
def test_non_orphan(self):
"""test that an entity can have two parent delete-orphan
cascades, and persists normally."""
class Address(_fixtures.Base):
pass
class Home(_fixtures.Base):
pass
class Business(_fixtures.Base):
pass
mapper(Address, addresses)
mapper(Home, homes, properties={'address'
: relationship(Address, cascade='all,delete-orphan',
single_parent=True)})
mapper(Business, businesses, properties={'address'
: relationship(Address, cascade='all,delete-orphan',
single_parent=True)})
session = create_session()
h1 = Home(description='home1', address=Address(street='address1'))
b1 = Business(description='business1',
address=Address(street='address2'))
session.add_all((h1,b1))
session.flush()
session.expunge_all()
eq_(session.query(Home).get(h1.id), Home(description='home1',
address=Address(street='address1')))
eq_(session.query(Business).get(b1.id),
Business(description='business1',
address=Address(street='address2')))
开发者ID:AndryulE,项目名称:kitsune,代码行数:32,代码来源:test_cascade.py
示例18: test_self_referential
def test_self_referential(self):
sess = create_session()
c1_employees = [e1, e2, b1, m1]
palias = aliased(Person)
eq_(
sess.query(Person, palias).filter(Person.company_id==palias.company_id).filter(Person.name=='dogbert').\
filter(Person.person_id>palias.person_id).order_by(Person.person_id, palias.person_id).all(),
[
(m1, e1),
(m1, e2),
(m1, b1),
]
)
eq_(
sess.query(Person, palias).filter(Person.company_id==palias.company_id).filter(Person.name=='dogbert').\
filter(Person.person_id>palias.person_id).from_self().order_by(Person.person_id, palias.person_id).all(),
[
(m1, e1),
(m1, e2),
(m1, b1),
]
)
开发者ID:gajop,项目名称:springgrid,代码行数:25,代码来源:test_query.py
示例19: testbasic
def testbasic(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
def __repr__(self):
return "Foo id %d, data %s" % (self.id, self.data)
mapper(Foo, foo)
class Bar(Foo):
def __repr__(self):
return "Bar id %d, data %s" % (self.id, self.data)
mapper(Bar, bar, inherits=Foo, properties={"foos": relation(Foo, secondary=bar_foo, lazy=True)})
sess = create_session()
b = Bar("bar #1")
sess.add(b)
b.foos.append(Foo("foo #1"))
b.foos.append(Foo("foo #2"))
sess.flush()
compare = repr(b) + repr(sorted([repr(o) for o in b.foos]))
sess.expunge_all()
l = sess.query(Bar).all()
print repr(l[0]) + repr(l[0].foos)
found = repr(l[0]) + repr(sorted([repr(o) for o in l[0].foos]))
eq_(found, compare)
开发者ID:hughperkins,项目名称:ailadder,代码行数:28,代码来源:test_manytomany.py
示例20: test_populate_dict
def test_populate_dict(self):
uowcommit, a1, b1, a_mapper, b_mapper = self._fixture()
a1.obj().id = 10
pairs = [(a_mapper.c.id, b_mapper.c.id,)]
dest = {}
sync.populate_dict(a1, a_mapper, dest, pairs)
eq_(dest, {'id':10})
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:7,代码来源:test_sync.py
注:本文中的sqlalchemy.test.testing.eq_函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论