• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python orm.dynamic_loader函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了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;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python orm.eagerload函数代码示例发布时间:2022-05-27
下一篇:
Python orm.deferred函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap