本文整理汇总了Python中sqlalchemy.util.pickle.loads函数的典型用法代码示例。如果您正苦于以下问题：Python loads函数的具体用法？Python loads怎么用？Python loads使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_class_deferred_cols
def test_class_deferred_cols(self):
addresses, users = (self.tables.addresses,
self.tables.users)
mapper(User, users, properties={
'name': sa.orm.deferred(users.c.name),
'addresses': relationship(Address, backref="user")
})
mapper(Address, addresses, properties={
'email_address': sa.orm.deferred(addresses.c.email_address)
})
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='[email protected]'))
sess.add(u1)
sess.flush()
sess.expunge_all()
u1 = sess.query(User).get(u1.id)
assert 'name' not in u1.__dict__
assert 'addresses' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.add(u2)
eq_(u2.name, 'ed')
eq_(u2, User(name='ed', addresses=[Address(email_address='[email protected]')]))
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, load=False)
eq_(u2.name, 'ed')
eq_(u2, User(name='ed', addresses=[Address(email_address='[email protected]')]))
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:32，代码来源:test_pickled.py
示例2: test_collection_setstate
def test_collection_setstate(self):
"""test a particular cycle that requires CollectionAdapter
to not rely upon InstanceState to deserialize."""
m = MetaData()
c1 = Table('c1', m,
Column('parent_id', String,
ForeignKey('p.id'), primary_key=True)
)
c2 = Table('c2', m,
Column('parent_id', String,
ForeignKey('p.id'), primary_key=True)
)
p = Table('p', m,
Column('id', String, primary_key=True)
)
mapper(Parent, p, properties={
'children1':relationship(Child1),
'children2':relationship(Child2)
})
mapper(Child1, c1)
mapper(Child2, c2)
obj = Parent()
screen1 = Screen(obj)
screen1.errors = [obj.children1, obj.children2]
screen2 = Screen(Child2(), screen1)
pickle.loads(pickle.dumps(screen2))
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:29，代码来源:test_pickled.py
示例3: test_options_with_descriptors
def test_options_with_descriptors(self):
users, addresses, dingalings = (self.tables.users,
self.tables.addresses,
self.tables.dingalings)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
})
mapper(Address, addresses, properties={
'dingaling':relationship(Dingaling)
})
mapper(Dingaling, dingalings)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='[email protected]'))
sess.add(u1)
sess.flush()
sess.expunge_all()
for opt in [
sa.orm.joinedload(User.addresses),
sa.orm.joinedload("addresses"),
sa.orm.defer("name"),
sa.orm.defer(User.name),
sa.orm.joinedload("addresses", Address.dingaling),
]:
opt2 = pickle.loads(pickle.dumps(opt))
eq_(opt.key, opt2.key)
u1 = sess.query(User).options(opt).first()
u2 = pickle.loads(pickle.dumps(u1))
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:31，代码来源:test_pickled.py
示例4: test_instance_deferred_cols
def test_instance_deferred_cols(self):
users, addresses = (self.tables.users, self.tables.addresses)
mapper(
User,
users,
properties={"addresses": relationship(Address, backref="user")},
)
mapper(Address, addresses)
sess = create_session()
u1 = User(name="ed")
u1.addresses.append(Address(email_address="[email protected]"))
sess.add(u1)
sess.flush()
sess.expunge_all()
u1 = (
sess.query(User)
.options(
sa.orm.defer("name"), sa.orm.defer("addresses.email_address")
)
.get(u1.id)
)
assert "name" not in u1.__dict__
assert "addresses" not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.add(u2)
eq_(u2.name, "ed")
assert "addresses" not in u2.__dict__
ad = u2.addresses[0]
assert "email_address" not in ad.__dict__
eq_(ad.email_address, "[email protected]")
eq_(
u2,
User(name="ed", addresses=[Address(email_address="[email protected]")]),
)
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, load=False)
eq_(u2.name, "ed")
assert "addresses" not in u2.__dict__
ad = u2.addresses[0]
# mapper options now transmit over merge(),
# new as of 0.6, so email_address is deferred.
assert "email_address" not in ad.__dict__
eq_(ad.email_address, "[email protected]")
eq_(
u2,
User(name="ed", addresses=[Address(email_address="[email protected]")]),
)
开发者ID:BY-jk，项目名称:sqlalchemy，代码行数:56，代码来源:test_pickled.py
示例5: test_polymorphic_deferred
def test_polymorphic_deferred(self):
email_users, users = (self.tables.email_users, self.tables.users)
mapper(
User,
users,
polymorphic_identity="user",
polymorphic_on=users.c.type,
)
mapper(
EmailUser,
email_users,
inherits=User,
polymorphic_identity="emailuser",
)
eu = EmailUser(name="user1", email_address="[email protected]")
sess = create_session()
sess.add(eu)
sess.flush()
sess.expunge_all()
eu = sess.query(User).first()
eu2 = pickle.loads(pickle.dumps(eu))
sess2 = create_session()
sess2.add(eu2)
assert "email_address" not in eu2.__dict__
eq_(eu2.email_address, "[email protected]")
开发者ID:BY-jk，项目名称:sqlalchemy，代码行数:28，代码来源:test_pickled.py
示例6: loads
def loads(cls, data):
try:
if isinstance(data, str):
data = data.encode()
return _phpsa.loads(data, decode_strings=True)
except:
return pickle.loads(data)
开发者ID:baloon11，项目名称:npui，代码行数:7，代码来源:phps.py
示例7: test_instance_lazy_relation_loaders
def test_instance_lazy_relation_loaders(self):
users, addresses = (self.tables.users,
self.tables.addresses)
mapper(User, users, properties={
'addresses': relationship(Address, lazy='noload')
})
mapper(Address, addresses)
sess = Session()
u1 = User(name='ed', addresses=[
Address(
email_address='[email protected]',
)
])
sess.add(u1)
sess.commit()
sess.close()
u1 = sess.query(User).options(
lazyload(User.addresses)
).first()
u2 = pickle.loads(pickle.dumps(u1))
sess = Session()
sess.add(u2)
assert u2.addresses
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:28，代码来源:test_pickled.py
示例8: test_weakref_pickled
def test_weakref_pickled(self):
users, User = self.tables.users, pickleable.User
s = create_session()
mapper(User, users)
s.add(User(name='ed'))
s.flush()
assert not s.dirty
user = s.query(User).one()
user.name = 'fred'
s.expunge(user)
u2 = pickle.loads(pickle.dumps(user))
del user
s.add(u2)
del u2
gc_collect()
assert len(s.identity_map) == 1
assert len(s.dirty) == 1
assert None not in s.dirty
s.flush()
gc_collect()
assert not s.dirty
assert not s.identity_map
开发者ID:DeepakAlevoor，项目名称:sqlalchemy，代码行数:30，代码来源:test_session.py
示例9: test_instance_deferred_cols
def test_instance_deferred_cols(self):
users, addresses = (self.tables.users, self.tables.addresses)
mapper(User, users, properties={
'addresses': relationship(Address, backref="user")
})
mapper(Address, addresses)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='[email protected]'))
sess.add(u1)
sess.flush()
sess.expunge_all()
u1 = sess.query(User).\
options(sa.orm.defer('name'),
sa.orm.defer('addresses.email_address')).\
get(u1.id)
assert 'name' not in u1.__dict__
assert 'addresses' not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.add(u2)
eq_(u2.name, 'ed')
assert 'addresses' not in u2.__dict__
ad = u2.addresses[0]
assert 'email_address' not in ad.__dict__
eq_(ad.email_address, '[email protected]')
eq_(u2, User(name='ed', addresses=[
Address(email_address='[email protected]')]))
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, load=False)
eq_(u2.name, 'ed')
assert 'addresses' not in u2.__dict__
ad = u2.addresses[0]
# mapper options now transmit over merge(),
# new as of 0.6, so email_address is deferred.
assert 'email_address' not in ad.__dict__
eq_(ad.email_address, '[email protected]')
eq_(u2, User(name='ed', addresses=[
Address(email_address='[email protected]')]))
开发者ID:zhsj，项目名称:sqlalchemy，代码行数:47，代码来源:test_pickled.py
示例10: test_load
def test_load(self):
s = Session()
with_poly = with_polymorphic(Person, [Engineer, Manager], flat=True)
emp = s.query(Company).options(
subqueryload(Company.employees.of_type(with_poly))).first()
e2 = pickle.loads(pickle.dumps(emp))
开发者ID:m32，项目名称:sqlalchemy，代码行数:8，代码来源:test_pickled.py
示例11: test_class_deferred_cols
def test_class_deferred_cols(self):
addresses, users = (self.tables.addresses, self.tables.users)
mapper(
User,
users,
properties={
"name": sa.orm.deferred(users.c.name),
"addresses": relationship(Address, backref="user"),
},
)
mapper(
Address,
addresses,
properties={
"email_address": sa.orm.deferred(addresses.c.email_address)
},
)
sess = create_session()
u1 = User(name="ed")
u1.addresses.append(Address(email_address="[email protected]"))
sess.add(u1)
sess.flush()
sess.expunge_all()
u1 = sess.query(User).get(u1.id)
assert "name" not in u1.__dict__
assert "addresses" not in u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
sess2.add(u2)
eq_(u2.name, "ed")
eq_(
u2,
User(name="ed", addresses=[Address(email_address="[email protected]")]),
)
u2 = pickle.loads(pickle.dumps(u1))
sess2 = create_session()
u2 = sess2.merge(u2, load=False)
eq_(u2.name, "ed")
eq_(
u2,
User(name="ed", addresses=[Address(email_address="[email protected]")]),
)
开发者ID:BY-jk，项目名称:sqlalchemy，代码行数:45，代码来源:test_pickled.py
示例12: test_pickled_update
def test_pickled_update(self):
users, User = self.tables.users, pickleable.User
mapper(User, users)
sess1 = create_session()
sess2 = create_session()
u1 = User(name="u1")
sess1.add(u1)
assert_raises_message(sa.exc.InvalidRequestError, "already attached to session", sess2.add, u1)
u2 = pickle.loads(pickle.dumps(u1))
sess2.add(u2)
开发者ID:vishvananda，项目名称:sqlalchemy，代码行数:11，代码来源:test_session.py
示例13: test_tuple_labeling
def test_tuple_labeling(self):
users = self.tables.users
sess = create_session()
# test pickle + all the protocols !
for pickled in False, -1, 0, 1, 2:
for row in sess.query(User, Address).join(User.addresses).all():
if pickled is not False:
row = pickle.loads(pickle.dumps(row, pickled))
eq_(list(row.keys()), ["User", "Address"])
eq_(row.User, row[0])
eq_(row.Address, row[1])
for row in sess.query(User.name, User.id.label("foobar")):
if pickled is not False:
row = pickle.loads(pickle.dumps(row, pickled))
eq_(list(row.keys()), ["name", "foobar"])
eq_(row.name, row[0])
eq_(row.foobar, row[1])
for row in sess.query(User).values(
User.name, User.id.label("foobar")
):
if pickled is not False:
row = pickle.loads(pickle.dumps(row, pickled))
eq_(list(row.keys()), ["name", "foobar"])
eq_(row.name, row[0])
eq_(row.foobar, row[1])
oalias = aliased(Order)
for row in sess.query(User, oalias).join(User.orders).all():
if pickled is not False:
row = pickle.loads(pickle.dumps(row, pickled))
eq_(list(row.keys()), ["User"])
eq_(row.User, row[0])
oalias = aliased(Order, name="orders")
for row in (
sess.query(User, oalias).join(oalias, User.orders).all()
):
if pickled is not False:
row = pickle.loads(pickle.dumps(row, pickled))
eq_(list(row.keys()), ["User", "orders"])
eq_(row.User, row[0])
eq_(row.orders, row[1])
# test here that first col is not labeled, only
# one name in keys, matches correctly
for row in sess.query(User.name + "hoho", User.name):
eq_(list(row.keys()), ["name"])
eq_(row[0], row.name + "hoho")
if pickled is not False:
ret = sess.query(User, Address).join(User.addresses).all()
pickle.loads(pickle.dumps(ret, pickled))
开发者ID:BY-jk，项目名称:sqlalchemy，代码行数:56，代码来源:test_pickled.py
示例14: test_backup_tables_table_with_multiple_blocks
def test_backup_tables_table_with_multiple_blocks(self):
tables={
'table1': MagicMock(**{'select.return_value': 'select from table1'})
}
self.dmp.info['meta']=MagicMock(tables=tables)
res1=MagicMock(**{'fetchmany.side_effect': ['block1','block2','block3',[]]})
self.dmp.con.execute=MagicMock(return_value=res1)
self.dmp.backup_tables()
self.dmp.con.execute.assert_called_once_with('select from table1')
with open(os.path.join(self.dmp.backup_dir,'table1.pickle'),'rb') as fh:
l=fh.readline()
self.assertEqual('block1',pickle.loads(fh.read(int(l))))
l=fh.readline()
self.assertEqual('block2',pickle.loads(fh.read(int(l))))
l=fh.readline()
self.assertEqual('block3',pickle.loads(fh.read(int(l))))
l=fh.readline()
self.assertEqual('EOF',l)
开发者ID:campenberger，项目名称:albackup，代码行数:22，代码来源:TestDump.py
示例15: test_invalidated_flag_pickle
def test_invalidated_flag_pickle(self):
users, addresses = (self.tables.users, self.tables.addresses)
mapper(User, users, properties={
'addresses': relationship(Address, lazy='noload')
})
mapper(Address, addresses)
u1 = User()
u1.addresses.append(Address())
u2 = pickle.loads(pickle.dumps(u1))
u2.addresses.append(Address())
eq_(len(u2.addresses), 2)
开发者ID:zhsj，项目名称:sqlalchemy，代码行数:13，代码来源:test_pickled.py
示例16: test_options_of_type
def test_options_of_type(self):
with_poly = with_polymorphic(Person, [Engineer, Manager], flat=True)
for opt, serialized in [
(
sa.orm.joinedload(Company.employees.of_type(Engineer)),
[(Company, "employees", Engineer)]),
(
sa.orm.joinedload(Company.employees.of_type(with_poly)),
[(Company, "employees", None)]),
]:
opt2 = pickle.loads(pickle.dumps(opt))
eq_(opt.__getstate__()['path'], serialized)
eq_(opt2.__getstate__()['path'], serialized)
开发者ID:m32，项目名称:sqlalchemy，代码行数:14，代码来源:test_pickled.py
示例17: persistent_load
def persistent_load(id):
m = our_ids.match(id)
if not m:
return None
else:
type_, args = m.group(1, 2)
if type_ == "attribute":
key, clsarg = args.split(":")
cls = pickle.loads(b64decode(clsarg))
return getattr(cls, key)
elif type_ == "mapper":
cls = pickle.loads(b64decode(args))
return class_mapper(cls)
elif type_ == "table":
return metadata.tables[args]
elif type_ == "column":
table, colname = args.split(":")
return metadata.tables[table].c[colname]
elif type_ == "session":
return scoped_session()
elif type_ == "engine":
return get_engine()
else:
raise Exception("Unknown token: %s" % type_)
开发者ID:lulzmachine，项目名称:rad，代码行数:24，代码来源:serializer.py
示例18: test_no_instrumentation
def test_no_instrumentation(self):
users = self.tables.users
umapper = mapper(User, users)
u1 = User(name='ed')
u1_pickled = pickle.dumps(u1, -1)
clear_mappers()
umapper = mapper(User, users)
u1 = pickle.loads(u1_pickled)
# this fails unless the InstanceState
# compiles the mapper
eq_(str(u1), "User(name='ed')")
开发者ID:zhsj，项目名称:sqlalchemy，代码行数:15，代码来源:test_pickled.py
示例19: test_rebuild_state
def test_rebuild_state(self):
"""not much of a 'test', but illustrate how to
remove instance-level state before pickling.
"""
users = self.tables.users
mapper(User, users)
u1 = User()
attributes.manager_of_class(User).teardown_instance(u1)
assert not u1.__dict__
u2 = pickle.loads(pickle.dumps(u1))
attributes.manager_of_class(User).setup_instance(u2)
assert attributes.instance_state(u2)
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:16，代码来源:test_pickled.py
示例20: test_transient
def test_transient(self):
users, addresses = (self.tables.users,
self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
})
mapper(Address, addresses)
sess = create_session()
u1 = User(name='ed')
u1.addresses.append(Address(email_address='[email protected]'))
u2 = pickle.loads(pickle.dumps(u1))
sess.add(u2)
sess.flush()
sess.expunge_all()
eq_(u1, sess.query(User).get(u2.id))
开发者ID:afeide，项目名称:LuoYunCloud，代码行数:20，代码来源:test_pickled.py
注：本文中的sqlalchemy.util.pickle.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论