本文整理汇总了Python中sqlalchemy.orm.dynamic_loader函数的典型用法代码示例。如果您正苦于以下问题:Python dynamic_loader函数的具体用法?Python dynamic_loader怎么用?Python dynamic_loader使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了dynamic_loader函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setup_model
def setup_model(model, metadata, **p):
model.UserBalance = UserBalance
model.users_balances_table = Table("users_balances", metadata, autoload=True)
model.users_balances_mapper = mapper(UserBalance, model.users_balances_table)
model.AuthenticatedUser = AuthenticatedUser
model.authenticated_users_table = Table("authenticated_users", metadata, autoload=True)
model.authenticated_users_mapper = mapper(AuthenticatedUser, model.authenticated_users_table,
properties = {
"balances": relation(model.UserBalance, lazy=True, backref="user", cascade="all, delete-orphan"),
})
model.PasswordResetRequest = PasswordResetRequest
model.password_reset_requests_table = Table("password_reset_requests", metadata, autoload=True)
model.password_reset_requests_mapper = mapper(PasswordResetRequest, model.password_reset_requests_table)
model.password_reset_requests_mapper.add_property("user", relation(model.AuthenticatedUser))
model.BalanceInvitation = BalanceInvitation
model.balance_invitations_table = Table("balance_invitations", metadata, autoload=True)
model.balance_invitations_mapper = mapper(BalanceInvitation, model.balance_invitations_table)
model.balance_invitations_mapper.add_property("sender", relation(model.AuthenticatedUser))
model.Invitation = Invitation
model.invitations_table = Table("invitations", metadata, autoload=True)
model.invitations_mapper = mapper(Invitation, model.invitations_table)
model.invitations_mapper.add_property("sender", relation(model.AuthenticatedUser))
model.BalanceTag = BalanceTag
model.balance_tags_table = Table("balance_tags", metadata, autoload=True)
model.balance_tags_mapper = mapper(BalanceTag, model.balance_tags_table)
model.ChangeTag = ChangeTag
model.changes_tags_table = Table("changes_tags", metadata, autoload=True)
model.changes_tags_mapper = mapper(ChangeTag, model.changes_tags_table)
model.BalanceChange = BalanceChange
model.balance_changes_table = Table("balance_changes", metadata, autoload=True)
model.balance_changes_mapper = mapper(BalanceChange, model.balance_changes_table,
properties = {
"tags": relation(model.BalanceTag, secondary=model.changes_tags_table, backref="changes"),
})
model.balance_changes_mapper.add_property("user", relation(model.AuthenticatedUser))
model.Balance = Balance
model.balances_table = Table("balances", metadata, autoload=True)
model.balances_mapper = mapper(Balance, model.balances_table,
properties = {
"users" : dynamic_loader(model.UserBalance, backref="balance", cascade="all, delete-orphan"),
"tags" : dynamic_loader(model.BalanceTag, backref="balance", cascade="all, delete-orphan"),
"invitations": dynamic_loader(model.BalanceInvitation, backref="balance", cascade="all, delete-orphan"),
"changes": dynamic_loader(model.BalanceChange, backref="balance", cascade="all, delete-orphan"),
})
model.ChangeCategory = ChangeCategory
model.change_categories_table = Table("change_categories", metadata, autoload=True)
model.change_categories_mapper = mapper(ChangeCategory, model.change_categories_table,
properties = {
"balance_changes" : dynamic_loader(model.BalanceChange, backref="change_category"),
})
开发者ID:pawelniewie,项目名称:5groszy.pl,代码行数:60,代码来源:__init__.py
示例2: setup_model
def setup_model(model, metadata):
"""
Define model used by our application - classes, tables, relationships.
"""
model.UserBalance = UserBalance
model.users_balances_table = Table("users_balances", metadata, autoload=True)
model.users_balances_mapper = mapper(UserBalance, model.users_balances_table)
model.AuthenticatedUser = AuthenticatedUser
model.authenticated_users_table = Table("authenticated_users", metadata, autoload=True)
model.authenticated_users_mapper = mapper(AuthenticatedUser, model.authenticated_users_table,
properties = {
"balances": relation(model.UserBalance, lazy=True, backref="user", cascade="all, delete-orphan"),
})
model.PasswordResetRequest = PasswordResetRequest
model.password_reset_requests_table = Table("password_reset_requests", metadata, autoload=True)
model.password_reset_requests_mapper = mapper(PasswordResetRequest, model.password_reset_requests_table)
model.password_reset_requests_mapper.add_property("user", relation(model.AuthenticatedUser))
model.BalanceInvitation = BalanceInvitation
model.balance_invitations_table = Table("balance_invitations", metadata, autoload=True)
model.balance_invitations_mapper = mapper(BalanceInvitation, model.balance_invitations_table)
model.balance_invitations_mapper.add_property("sender", relation(model.AuthenticatedUser))
model.Invitation = Invitation
model.invitations_table = Table("invitations", metadata, autoload=True)
model.invitations_mapper = mapper(Invitation, model.invitations_table)
model.invitations_mapper.add_property("sender", relation(model.AuthenticatedUser))
model.ChangeTag = ChangeTag
model.changes_tags_table = Table("changes_tags", metadata, autoload=True)
model.changes_tags_mapper = mapper(ChangeTag, model.changes_tags_table)
model.BalanceChange = BalanceChange
model.balance_changes_table = Table("balance_changes", metadata, autoload=True)
model.balance_changes_mapper = mapper(BalanceChange, model.balance_changes_table,
properties = {
"tags": relation(model.ChangeTag, backref="change", cascade="all, delete-orphan", lazy=False),
"transfer": relation(model.BalanceChange, uselist=False, remote_side=[model.balance_changes_table.c.uid], lazy=False, post_update=True),
})
model.balance_changes_mapper.add_property("user", relation(model.AuthenticatedUser))
model.Balance = Balance
model.balances_table = Table("balances", metadata, autoload=True)
model.balances_mapper = mapper(Balance, model.balances_table,
properties = {
"users" : dynamic_loader(model.UserBalance, backref="balance", cascade="all, delete-orphan"),
"invitations": dynamic_loader(model.BalanceInvitation, backref="balance", cascade="all, delete-orphan"),
"changes": dynamic_loader(model.BalanceChange, backref="balance", cascade="all, delete-orphan"),
})
model.ChangeCategory = ChangeCategory
model.change_categories_table = Table("change_categories", metadata, autoload=True)
model.change_categories_mapper = mapper(ChangeCategory, model.change_categories_table,
properties = {
"balance_changes" : dynamic_loader(model.BalanceChange, backref="change_category"),
})
开发者ID:pawelniewie,项目名称:5groszy.pl,代码行数:58,代码来源:__init__.py
示例3: test_delete_cascade
def test_delete_cascade(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
backref='user', cascade="all, delete-orphan")
})
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='c'), u.addresses[2])
sess.delete(u.addresses[2])
sess.delete(u.addresses[4])
sess.delete(u.addresses[3])
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
list(u.addresses))
sess.expunge_all()
u = sess.query(User).get(u.id)
sess.delete(u)
# u.addresses relationship will have to force the load
# of all addresses so that they can be updated
sess.flush()
sess.close()
eq_(testing.db.scalar(addresses.count()), 0)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:33,代码来源:test_dynamic.py
示例4: _backref_test
def _backref_test(self, autoflush, saveuser):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
})
sess = create_session(autoflush=autoflush)
u = User(name='buffy')
a = Address(email_address='[email protected]')
a.user = u
if saveuser:
sess.add(u)
else:
sess.add(a)
if not autoflush:
sess.flush()
assert u in sess
assert a in sess
eq_(list(u.addresses), [a])
a.user = None
if not autoflush:
eq_(list(u.addresses), [a])
if not autoflush:
sess.flush()
eq_(list(u.addresses), [])
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:31,代码来源:test_dynamic.py
示例5: test_flush
def test_flush(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u1 = User(name='jack')
u2 = User(name='ed')
u2.addresses.append(Address(email_address='[email protected]'))
u1.addresses.append(Address(email_address='[email protected]'))
sess.add_all((u1, u2))
sess.flush()
from sqlalchemy.orm import attributes
eq_(attributes.get_history(u1, 'addresses'), ([], [Address(email_address='[email protected]')], []))
sess.expunge_all()
# test the test fixture a little bit
ne_(User(name='jack', addresses=[Address(email_address='wrong')]),
sess.query(User).first())
eq_(User(name='jack', addresses=[Address(email_address='[email protected]')]),
sess.query(User).first())
eq_([
User(name='jack', addresses=[Address(email_address='[email protected]')]),
User(name='ed', addresses=[Address(email_address='[email protected]')])
],
sess.query(User).all())
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:28,代码来源:test_dynamic.py
示例6: test_merge
def test_merge(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address)
})
sess = create_session()
u1 = User(name='jack')
a1 = Address(email_address='a1')
a2 = Address(email_address='a2')
a3 = Address(email_address='a3')
u1.addresses.append(a2)
u1.addresses.append(a3)
sess.add_all([u1, a1])
sess.flush()
u1 = User(id=u1.id, name='jack')
u1.addresses.append(a1)
u1.addresses.append(a3)
u1 = sess.merge(u1)
eq_(attributes.get_history(u1, 'addresses'), (
[a1],
[a3],
[a2]
))
sess.flush()
eq_(
list(u1.addresses),
[a1, a3]
)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:32,代码来源:test_dynamic.py
示例7: 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
示例8: test_delete_nocascade
def test_delete_nocascade(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id, 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)
assert Address(email_address='c') == u.addresses[2]
sess.delete(u.addresses[2])
sess.delete(u.addresses[4])
sess.delete(u.addresses[3])
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
sess.clear()
u = sess.query(User).get(u.id)
sess.delete(u)
# u.addresses relation will have to force the load
# of all addresses so that they can be updated
sess.flush()
sess.close()
assert testing.db.scalar(addresses.count(addresses.c.user_id != None)) ==0
开发者ID:jrus,项目名称:sqlalchemy,代码行数:31,代码来源:dynamic.py
示例9: test_configured_order_by
def test_configured_order_by(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), order_by=desc(Address.email_address))
})
sess = create_session()
u = sess.query(User).get(8)
eq_(list(u.addresses), [Address(email_address=u'[email protected]'), Address(email_address=u'[email protected]'), Address(email_address=u'[email protected]')])
开发者ID:jrus,项目名称:sqlalchemy,代码行数:7,代码来源:dynamic.py
示例10: test_basic
def test_basic(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u1 = User(name='jack')
u2 = User(name='ed')
u2.addresses.append(Address(email_address='[email protected]'))
u1.addresses.append(Address(email_address='[email protected]'))
sess.add_all((u1, u2))
sess.flush()
from sqlalchemy.orm import attributes
self.assertEquals(attributes.get_history(attributes.instance_state(u1), 'addresses'), ([], [Address(email_address='[email protected]')], []))
sess.clear()
# test the test fixture a little bit
assert User(name='jack', addresses=[Address(email_address='wrong')]) != sess.query(User).first()
assert User(name='jack', addresses=[Address(email_address='[email protected]')]) == sess.query(User).first()
assert [
User(name='jack', addresses=[Address(email_address='[email protected]')]),
User(name='ed', addresses=[Address(email_address='[email protected]')])
] == sess.query(User).all()
开发者ID:jrus,项目名称:sqlalchemy,代码行数:25,代码来源:dynamic.py
示例11: test_events
def test_events(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u1 = User(name='jack')
a1 = Address(email_address='foo')
sess.add_all([u1, a1])
sess.flush()
eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
u1 = sess.query(User).get(u1.id)
u1.addresses.append(a1)
sess.flush()
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
[(a1.id, u1.id, 'foo')])
u1.addresses.remove(a1)
sess.flush()
eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
u1.addresses.append(a1)
sess.flush()
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
[(a1.id, u1.id, 'foo')])
a2 = Address(email_address='bar')
u1.addresses.remove(a1)
u1.addresses.append(a2)
sess.flush()
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
[(a2.id, u1.id, 'bar')])
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:33,代码来源:test_dynamic.py
示例12: 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
示例13: test_transient_detached
def test_transient_detached(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u1 = User()
u1.addresses.append(Address())
eq_(u1.addresses.count(), 1)
eq_(u1.addresses[0], Address())
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:9,代码来源:test_dynamic.py
示例14: test_hasattr
def test_hasattr(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
u1 = User(name='jack')
assert 'addresses' not in u1.__dict__.keys()
u1.addresses = [Address(email_address='test')]
assert 'addresses' in dir(u1)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:9,代码来源:test_dynamic.py
示例15: test_no_populate
def test_no_populate(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
u1 = User()
assert_raises_message(
NotImplementedError,
"Dynamic attributes don't support collection population.",
attributes.set_committed_value, u1, 'addresses', []
)
开发者ID:AndryulE,项目名称:kitsune,代码行数:10,代码来源:test_dynamic.py
示例16: test_basic
def test_basic(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
q = sess.query(User)
u = q.filter(User.id==7).first()
eq_([User(id=7,
addresses=[Address(id=1, email_address='[email protected]')])],
q.filter(User.id==7).all())
eq_(self.static.user_address_result, q.all())
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:12,代码来源:test_dynamic.py
示例17: test_rollback
def test_rollback(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session(expire_on_commit=False, autocommit=False, autoflush=True)
u1 = User(name='jack')
u1.addresses.append(Address(email_address='[email protected]'))
sess.add(u1)
sess.flush()
sess.commit()
u1.addresses.append(Address(email_address='[email protected]'))
eq_(u1.addresses.all(), [Address(email_address='[email protected]'), Address(email_address='[email protected]')])
sess.rollback()
eq_(u1.addresses.all(), [Address(email_address='[email protected]')])
开发者ID:jrus,项目名称:sqlalchemy,代码行数:14,代码来源:dynamic.py
示例18: config_keyable
def config_keyable(cls,name="settings"):
"""Object - List Mixin/Mutator"""
mapper = class_mapper(cls)
table = mapper.local_table
cls_name = str(cls)
cls_name = cls_name[cls_name.rfind('.')+1:cls_name.rfind('\'')].lower()
mapper.add_property(name, dynamic_loader(ConfigAttribute,
primaryjoin=and_(table.c.id==configattribute_table.c.object_id,configattribute_table.c.object_type==cls_name),
foreign_keys=[configattribute_table.c.object_id],
backref='%s' % table.name))
# initialize some stuff
def on_new(self):
self.object_type = cls_name
setattr(cls, "on_new", on_new)
开发者ID:araddon,项目名称:demisauce,代码行数:15,代码来源:config.py
示例19: test_no_count
def test_no_count(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
q = sess.query(User)
# dynamic collection cannot implement __len__() (at least one that
# returns a live database result), else additional count() queries are
# issued when evaluating in a list context
def go():
eq_([User(id=7,
addresses=[Address(id=1,
email_address='[email protected]')])],
q.filter(User.id==7).all())
self.assert_sql_count(testing.db, go, 2)
开发者ID:gaguilarmi,项目名称:sqlalchemy,代码行数:16,代码来源:test_dynamic.py
示例20: userlistable
def userlistable(cls,name="users"):
"""User - List Mixin/Mutator"""
mapper = class_mapper(cls)
table = mapper.local_table
cls_name = str(cls)
cls_name = cls_name[cls_name.rfind('.')+1:cls_name.rfind('\'')].lower()
mapper.add_property(name, dynamic_loader(UserAttribute,
primaryjoin=and_(table.c.id==userattribute_table.c.object_id,userattribute_table.c.object_type==cls_name),
foreign_keys=[userattribute_table.c.object_id],
backref='%s' % table.name))
#log.debug("userlistable table.name = %s" % table.name)
# initialize some stuff
def on_new(self):
self.object_type = cls_name
setattr(cls, "on_new", on_new)
开发者ID:araddon,项目名称:demisauce,代码行数:16,代码来源:user.py
注:本文中的sqlalchemy.orm.dynamic_loader函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论