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

Python orm.deferred函数代码示例

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

本文整理汇总了Python中sqlalchemy.orm.deferred函数的典型用法代码示例。如果您正苦于以下问题:Python deferred函数的具体用法?Python deferred怎么用?Python deferred使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了deferred函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_commits_state

    def test_commits_state(self):
        """
        When deferred elements are loaded via a group, they get the proper
        CommittedState and don't result in changes being committed

        """

        orders, Order = self.tables.orders, self.classes.Order

        mapper(
            Order,
            orders,
            properties={
                "userident": deferred(orders.c.user_id, group="primary"),
                "description": deferred(orders.c.description, group="primary"),
                "opened": deferred(orders.c.isopen, group="primary"),
            },
        )

        sess = create_session()
        o2 = sess.query(Order).get(3)

        # this will load the group of attributes
        eq_(o2.description, "order 3")
        assert o2 not in sess.dirty
        # this will mark it as 'dirty', but nothing actually changed
        o2.description = "order 3"
        # therefore the flush() shouldnt actually issue any SQL
        self.assert_sql_count(testing.db, sess.flush, 0)
开发者ID:niaolianyu,项目名称:sqlalchemy,代码行数:29,代码来源:test_deferred.py


示例2: test_preserve_changes

    def test_preserve_changes(self):
        """A deferred load operation doesn't revert modifications on attributes"""

        orders, Order = self.tables.orders, self.classes.Order

        mapper(
            Order,
            orders,
            properties={
                "userident": deferred(orders.c.user_id, group="primary"),
                "description": deferred(orders.c.description, group="primary"),
                "opened": deferred(orders.c.isopen, group="primary"),
            },
        )
        sess = create_session()
        o = sess.query(Order).get(3)
        assert "userident" not in o.__dict__
        o.description = "somenewdescription"
        eq_(o.description, "somenewdescription")

        def go():
            eq_(o.opened, 1)

        self.assert_sql_count(testing.db, go, 1)
        eq_(o.description, "somenewdescription")
        assert o in sess.dirty
开发者ID:niaolianyu,项目名称:sqlalchemy,代码行数:26,代码来源:test_deferred.py


示例3: test_undefer_group

    def test_undefer_group(self):
        orders, Order = self.tables.orders, self.classes.Order

        mapper(Order, orders, properties=util.OrderedDict([
            ('userident', deferred(orders.c.user_id, group='primary')),
            ('description', deferred(orders.c.description, group='primary')),
            ('opened', deferred(orders.c.isopen, group='primary'))
            ]
            ))

        sess = create_session()
        q = sess.query(Order).order_by(Order.id)
        def go():
            l = q.options(undefer_group('primary')).all()
            o2 = l[2]
            eq_(o2.opened, 1)
            eq_(o2.userident, 7)
            eq_(o2.description, 'order 3')

        self.sql_eq_(go, [
            ("SELECT orders.user_id AS orders_user_id, "
             "orders.description AS orders_description, "
             "orders.isopen AS orders_isopen, "
             "orders.id AS orders_id, "
             "orders.address_id AS orders_address_id "
             "FROM orders ORDER BY orders.id",
             {})])
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:27,代码来源:test_deferred.py


示例4: test_undefer_star

    def test_undefer_star(self):
        orders, Order = self.tables.orders, self.classes.Order

        mapper(
            Order,
            orders,
            properties=util.OrderedDict(
                [
                    ("userident", deferred(orders.c.user_id)),
                    ("description", deferred(orders.c.description)),
                    ("opened", deferred(orders.c.isopen)),
                ]
            ),
        )

        sess = create_session()
        q = sess.query(Order).options(Load(Order).undefer("*"))
        self.assert_compile(
            q,
            "SELECT orders.user_id AS orders_user_id, "
            "orders.description AS orders_description, "
            "orders.isopen AS orders_isopen, "
            "orders.id AS orders_id, "
            "orders.address_id AS orders_address_id FROM orders",
        )
开发者ID:niaolianyu,项目名称:sqlalchemy,代码行数:25,代码来源:test_deferred.py


示例5: test_group

    def test_group(self):
        """Deferred load with a group"""

        orders, Order = self.tables.orders, self.classes.Order

        mapper(
            Order,
            orders,
            properties=util.OrderedDict(
                [
                    ("userident", deferred(orders.c.user_id, group="primary")),
                    ("addrident", deferred(orders.c.address_id, group="primary")),
                    ("description", deferred(orders.c.description, group="primary")),
                    ("opened", deferred(orders.c.isopen, group="primary")),
                ]
            ),
        )

        sess = create_session()
        q = sess.query(Order).order_by(Order.id)

        def go():
            l = q.all()
            o2 = l[2]
            eq_(o2.opened, 1)
            eq_(o2.userident, 7)
            eq_(o2.description, "order 3")

        self.sql_eq_(
            go,
            [
                ("SELECT orders.id AS orders_id " "FROM orders ORDER BY orders.id", {}),
                (
                    "SELECT orders.user_id AS orders_user_id, "
                    "orders.address_id AS orders_address_id, "
                    "orders.description AS orders_description, "
                    "orders.isopen AS orders_isopen "
                    "FROM orders WHERE orders.id = :param_1",
                    {"param_1": 3},
                ),
            ],
        )

        o2 = q.all()[2]
        eq_(o2.description, "order 3")
        assert o2 not in sess.dirty
        o2.description = "order 3"

        def go():
            sess.flush()

        self.sql_count_(0, go)
开发者ID:niaolianyu,项目名称:sqlalchemy,代码行数:52,代码来源:test_deferred.py


示例6: test_unsaved_group_2

    def test_unsaved_group_2(self):
        orders, Order = self.tables.orders, self.classes.Order

        mapper(Order, orders, order_by=orders.c.id, properties=dict(
            description=deferred(orders.c.description, group='primary'),
            opened=deferred(orders.c.isopen, group='primary')))

        sess = create_session()
        o = Order()
        sess.add(o)
        def go():
            o.description = "some description"
        self.sql_count_(0, go)
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:13,代码来源:test_deferred.py


示例7: mapAll

def mapAll(metadata , allclasses):
    
    clear_mappers()
    
    for tablename, oeclass in allclasses.iteritems():
        table = metadata.tables[tablename]
        for parentname in oeclass.parents :
            table.c['id_'+parentname].append_foreign_key( ForeignKey(parentname+'.id') ) 
    
    
    
    for tablename, oeclass in allclasses.iteritems():
        table = metadata.tables[tablename]
        properties = { }
        for fieldname, fieldtype in oeclass.fields :
            if fieldtype == numpy.ndarray :
                properties[fieldname+'_shape'] = deferred( table.columns[fieldname+'_shape'] , group = fieldname)
                properties[fieldname+'_dtype'] = deferred( table.columns[fieldname+'_dtype'] , group = fieldname)
                properties[fieldname+'_blob'] = deferred( table.columns[fieldname+'_blob'] , group = fieldname)
        
        for child in oeclass.children :
            #~ properties['_'+child+'s'] = relationship(allclasses[child] , )
            #~ print tablename , child
            properties['_'+child+'s'] = relationship(allclasses[child] ,
                            primaryjoin = table.c.id==metadata.tables[child].c['id_'+tablename], 
                            order_by = metadata.tables[child].c['id'],
                            
                            
                            backref=backref(tablename),
                            
                            # FIXME
                            #~ cascade="all, delete, delete-orphan",
                            #~ cascade="all, delete, delete-orphan",
                            cascade="all, delete",
                            #~ cascade="all",
                            
                            #~ lazy = True,
                                        )
        
        mapper(oeclass , table , properties = properties , )
        
         #non_primary=True to create a non primary Mapper.  clear_mappers()
        
    # set numpy.ndarray field property for all classes
    for tablename, oeclass in allclasses.iteritems():
            for fieldname, fieldtype in oeclass.fields :
                if fieldtype == numpy.ndarray :
                    setattr(oeclass, fieldname, property( NumpyField(fieldname).getfield , NumpyField(fieldname).setfield) )
开发者ID:AntoineValera,项目名称:SynaptiQs,代码行数:48,代码来源:sqlalchemyutil.py


示例8: Doc

def Doc(models):

    if models.lang == 'ru':
        id = Column(Integer, primary_key=True)
    else:
        id = Column(Integer, ForeignKey(models.DocRu.id),
                    primary_key=True, autoincrement=False)

    date = Column(DateTime, nullable=False, default=datetime.now, index=True)
    title = Column(Html(String(1000)), nullable=False, default='')

    summary = Column(Html(Text), nullable=False, default='')
    body = deferred(Column(ExpandableHtml(MediumText), nullable=False,
                    default=ExpandableMarkup('')))

    _photos, photos_edit, photos = editable_ordered_relation(
            models.Doc_Photo, 'photo', use_property=False)
    _photo_sets, photo_sets_edit, photo_sets = editable_ordered_relation(
            models.Doc_PhotoSet, 'photo_set', use_property=False)

    link_blocks_edit = relationship(
            models.DocLinkBlock,
            order_by=[models.DocLinkBlock.order_position],
            collection_class=ordering_list('order_position'),
            cascade='all, delete-orphan')
    # do not display blocks without links
    link_blocks = FilteredProperty('link_blocks_edit', has_links=True)

    sections = relationship(
            models.Section,
            secondary=models.Doc_Section.__table__)

    __mapper_args__ = {'order_by': desc(date)}

    def __unicode__(self):
        if self.id is None:
            return u'Новый материал'
        return u'Материал: {}'.format(self.title)

    @cached_property
    def index_photo(self):
        if self.photos:
            return self.photos[0]
        elif self.photo_sets:
            return self.photo_sets[0].index_photo
        else:
            return None

    @cached_property
    def all_photos(self):
        photos = sum([x.photos for x in self.photo_sets], []) + self.photos
        return list(collections.OrderedDict.fromkeys(photos))

    @cached_property
    def links_count(self):
        return sum([len(x.links) for x in self.link_blocks])

    @cached_property
    def date_formatted(self):
        return format_datetime(self.date, locale=self.models.lang)
开发者ID:SmartTeleMax,项目名称:iktomi-cms-demo,代码行数:60,代码来源:docs.py


示例9: test_deep_options

    def test_deep_options(self):
        users, items, order_items, Order, Item, User, orders = (self.tables.users,
                                self.tables.items,
                                self.tables.order_items,
                                self.classes.Order,
                                self.classes.Item,
                                self.classes.User,
                                self.tables.orders)

        mapper(Item, items, properties=dict(
            description=deferred(items.c.description)))
        mapper(Order, orders, properties=dict(
            items=relationship(Item, secondary=order_items)))
        mapper(User, users, properties=dict(
            orders=relationship(Order, order_by=orders.c.id)))

        sess = create_session()
        q = sess.query(User).order_by(User.id)
        l = q.all()
        item = l[0].orders[1].items[1]
        def go():
            eq_(item.description, 'item 4')
        self.sql_count_(1, go)
        eq_(item.description, 'item 4')

        sess.expunge_all()
        l = q.options(undefer('orders.items.description')).all()
        item = l[0].orders[1].items[1]
        def go():
            eq_(item.description, 'item 4')
        self.sql_count_(0, go)
        eq_(item.description, 'item 4')
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:32,代码来源:test_deferred.py


示例10: sessionmaker

def sessionmaker(dbconfig):
    dbconfig = dbconfig.copy()
    conn_str = dbconfig.pop('url')
    if 'schema' in dbconfig:
        schema = dbconfig.pop('schema')
    else:
        schema = None

    engine = create_engine(conn_str, **dbconfig)
    mappers, tables, Session = reflect(engine, models, schema)

    # add mapper relationships
    mappers['Data'].add_properties({
        'versions': relationship(models.Version,
                                 lazy='dynamic',
                                 backref=backref('ref',
                                                 lazy='joined'))
    })

    mappers['Version'].add_properties({
        'data': deferred(tables['version'].c['data']),
        'size': column_property(func.length(tables['version'].c['data']))
    })

    Session.class_.mappers = mappers

    return Session
开发者ID:petrushev,项目名称:mkopen,代码行数:27,代码来源:mappers.py


示例11: modified

 def modified(cls):
     return deferred(
         Column(
             UTCDateTime(timezone=False),
             onupdate=cls.timestamp
         )
     )
开发者ID:seantis,项目名称:libres,代码行数:7,代码来源:timestamp.py


示例12: test_basic

    def test_basic(self):
        """A basic deferred load."""

        Order, orders = self.classes.Order, self.tables.orders


        mapper(Order, orders, order_by=orders.c.id, properties={
            'description': deferred(orders.c.description)})

        o = Order()
        self.assert_(o.description is None)

        q = create_session().query(Order)
        def go():
            l = q.all()
            o2 = l[2]
            x = o2.description

        self.sql_eq_(go, [
            ("SELECT orders.id AS orders_id, "
             "orders.user_id AS orders_user_id, "
             "orders.address_id AS orders_address_id, "
             "orders.isopen AS orders_isopen "
             "FROM orders ORDER BY orders.id", {}),
            ("SELECT orders.description AS orders_description "
             "FROM orders WHERE orders.id = :param_1",
             {'param_1':3})])
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:27,代码来源:test_deferred.py


示例13: created

 def created(cls):
     return deferred(
         Column(
             types.DateTime(timezone=True),
             default=utils.utcnow
         )
     )
开发者ID:4teamwork,项目名称:seantis.reservation,代码行数:7,代码来源:timestamp.py


示例14: modified

 def modified(cls):
     return deferred(
         Column(
             types.DateTime(timezone=True),
             onupdate=utils.utcnow
         )
     )
开发者ID:4teamwork,项目名称:seantis.reservation,代码行数:7,代码来源:timestamp.py


示例15: test_locates_col_rowproc_only

    def test_locates_col_rowproc_only(self):
        """changed in 1.0 - we don't search for deferred cols in the result
        now.

        Because the loading for ORM Query and Query from a core select
        is now split off, we test loading from a plain select()
        separately.

        """

        orders, Order = self.tables.orders, self.classes.Order


        mapper(Order, orders, properties={
            'description': deferred(orders.c.description)})

        sess = create_session()
        stmt = sa.select([Order]).order_by(Order.id)
        o1 = (sess.query(Order).
              from_statement(stmt).all())[0]
        def go():
            eq_(o1.description, 'order 1')
        # prior to 1.0 we'd search in the result for this column
        # self.sql_count_(0, go)
        self.sql_count_(1, go)
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:25,代码来源:test_deferred.py


示例16: make_deferred_properties

def make_deferred_properties(columns, defer_group, except_column_names):
    """Make a deferred group covering all columns except those specified.

       SQLAlchemy has a 'deferred' feature that allows you to avoid loading
       large infrequently-used columns until you explicitly access them.
       Typically the deferred columns appear only on detail pages, while the
       underferred columns also appear in indexes and simple searches.
       SQLAlchemy normally requires you to list deferred columns explicitly.
       This convenience function builds the list for you, deferring all
       columns not listed as undeferred.

       'columns': pass the .columns attribute from a SQLAlchemy Table.
       'defer_group' is the name of the defer group to create.
       'except_column_names' is a list of column names not to defer.

       Usage:

           _properties = make_deferred_properties(t_mytable.columns, "details",
               ["id", "title", "author"])
           sqlalchemy.orm.mapper(MyClass, t_mytable, properties=_properties)

           # Example query
           q = Session.query(MyClass)
           if details:
                q = q.option(sqlalchemy.orm.undefer_group("details"))
           records = q.all()
    """
    ret = {}
    for col in columns:
        if col.name not in except_column_names:
            ret[col.name] = orm.deferred(col, group=defer_group)
    return ret
开发者ID:JamesMakela-NOAA,项目名称:PyGnome,代码行数:32,代码来源:sqlalchemy_util.py


示例17: test_state_deferred_to_col

    def test_state_deferred_to_col(self):
        """Behavioral test to verify the current activity of loader callables."""

        users, User = self.tables.users, self.classes.User

        mapper(User, users, properties={"name": deferred(users.c.name)})

        sess = create_session()
        u1 = sess.query(User).options(undefer(User.name)).first()
        assert "name" not in attributes.instance_state(u1).callables

        # mass expire, the attribute was loaded,
        # the attribute gets the callable
        sess.expire(u1)
        assert isinstance(attributes.instance_state(u1).callables["name"], state.InstanceState)

        # load it, callable is gone
        u1.name
        assert "name" not in attributes.instance_state(u1).callables

        # mass expire, attribute was loaded but then deleted,
        # the callable goes away - the state wants to flip
        # it back to its "deferred" loader.
        sess.expunge_all()
        u1 = sess.query(User).options(undefer(User.name)).first()
        del u1.name
        sess.expire(u1)
        assert "name" not in attributes.instance_state(u1).callables

        # single attribute expire, the attribute gets the callable
        sess.expunge_all()
        u1 = sess.query(User).options(undefer(User.name)).first()
        sess.expire(u1, ["name"])
        assert isinstance(attributes.instance_state(u1).callables["name"], state.InstanceState)
开发者ID:onetera,项目名称:scandatatransfer,代码行数:34,代码来源:test_expire.py


示例18: created

 def created(cls):
     return deferred(
         Column(
             UTCDateTime(timezone=False),
             default=cls.timestamp
         )
     )
开发者ID:seantis,项目名称:libres,代码行数:7,代码来源:timestamp.py


示例19: make_table

def make_table(name, columns, base=None, **table_args):
    """Generate an ORM mapping class from a simplified schema format.

    Columns named 'id' (int) and 'meta' (object) are added automatically.

    Parameters
    ----------
    name : str
        Name of the table, used to set __tablename__ in the new class
    base : class or None
        Base class on which to build the new table class
    table_args : keyword arguments
        Extra keyword arguments are used to set __table_args__ in the new class
    columns : list of tuple
        List of column specifications. Each column is given as a tuple:
        ``(col_name, data_type, comment, {options})``. Where *col_name* and *comment* 
        are strings, *data_type* is a key in the column_data_types global, and
        *options* is a dict providing extra initialization arguments to the sqlalchemy
        Column (for example: 'index', 'unique'). Optionally, *data_type* may be a 'tablename.id'
        string indicating that this column is a foreign key referencing another table.
    """
    props = {
        '__tablename__': name,
        '__table_args__': table_args,
        'id': Column(Integer, primary_key=True),
    }

    for column in columns:
        colname, coltype = column[:2]
        kwds = {} if len(column) < 4 else column[3]
        kwds['comment'] = None if len(column) < 3 else column[2]
        defer_col = kwds.pop('deferred', False)
        ondelete = kwds.pop('ondelete', None)

        if coltype not in column_data_types:
            if not coltype.endswith('.id'):
                raise ValueError("Unrecognized column type %s" % coltype)
            props[colname] = Column(Integer, ForeignKey(coltype, ondelete=ondelete), **kwds)
        else:
            ctyp = column_data_types[coltype]
            props[colname] = Column(ctyp, **kwds)

        if defer_col:
            props[colname] = deferred(props[colname])

    # props['time_created'] = Column(DateTime, default=func.now())
    # props['time_modified'] = Column(DateTime, onupdate=func.current_timestamp())
    props['meta'] = Column(column_data_types['object'])

    if base is None:
        return type(name, (ORMBase,), props)
    else:
        # need to jump through a hoop to allow __init__ on table classes;
        # see: https://docs.sqlalchemy.org/en/latest/orm/constructors.html
        if hasattr(base, '_init_on_load'):
            @reconstructor
            def _init_on_load(self, *args, **kwds):
                base._init_on_load(self)
            props['_init_on_load'] = _init_on_load
        return type(name, (base,ORMBase), props)
开发者ID:corinneteeter,项目名称:multipatch_analysis,代码行数:60,代码来源:database.py


示例20: is_updated

 def is_updated(self):
     return deferred(
         sa.Column(
             sa.TIMESTAMP, nullable=False, default=datetime.datetime.now,
             server_default=sqlaexp.text('0'),
             onupdate=datetime.datetime.now,
             server_onupdate=sqlafunc.current_timestamp(),
             ))
开发者ID:TakesxiSximada,项目名称:azoth,代码行数:8,代码来源:models.py



注:本文中的sqlalchemy.orm.deferred函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python orm.dynamic_loader函数代码示例发布时间:2022-05-27
下一篇:
Python orm.defer函数代码示例发布时间: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