本文整理汇总了Python中sqlalchemy.orm.attributes.instance_dict函数的典型用法代码示例。如果您正苦于以下问题:Python instance_dict函数的具体用法?Python instance_dict怎么用?Python instance_dict使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了instance_dict函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_deferred
def test_deferred(self):
for base in (object, MyBaseClass, MyClass):
class Foo(base):
pass
data = {'a': 'this is a', 'b': 12}
def loader(state, keys):
for k in keys:
state.dict[k] = data[k]
return attributes.ATTR_WAS_SET
manager = register_class(Foo)
manager.deferred_scalar_loader = loader
attributes.register_attribute(
Foo, 'a', uselist=False, useobject=False)
attributes.register_attribute(
Foo, 'b', uselist=False, useobject=False)
if base is object:
assert Foo not in \
instrumentation._instrumentation_factory._state_finders
else:
assert Foo in \
instrumentation._instrumentation_factory._state_finders
f = Foo()
attributes.instance_state(f)._expire(
attributes.instance_dict(f), set())
eq_(f.a, "this is a")
eq_(f.b, 12)
f.a = "this is some new a"
attributes.instance_state(f)._expire(
attributes.instance_dict(f), set())
eq_(f.a, "this is a")
eq_(f.b, 12)
attributes.instance_state(f)._expire(
attributes.instance_dict(f), set())
f.a = "this is another new a"
eq_(f.a, "this is another new a")
eq_(f.b, 12)
attributes.instance_state(f)._expire(
attributes.instance_dict(f), set())
eq_(f.a, "this is a")
eq_(f.b, 12)
del f.a
eq_(f.a, None)
eq_(f.b, 12)
attributes.instance_state(f)._commit_all(
attributes.instance_dict(f))
eq_(f.a, None)
eq_(f.b, 12)
开发者ID:anti-social,项目名称:sqlalchemy,代码行数:57,代码来源:test_extendedattr.py
示例2: getMessageByHash
def getMessageByHash(self, message_hash):
"""Get a message for a given hash."""
try:
result = self.session.query(Messages).filter(Messages.message_hash == message_hash).one()
return instance_dict(result)
except NoResultFound, e:
return None
开发者ID:zeitkunst,项目名称:FluidNexus,代码行数:7,代码来源:Database.py
示例3: replicate_relation
def replicate_relation(source, target, attr, target_attr, cache=None):
if attr.property.cascade.delete_orphan:
process_scalar = replicate_no_merge
process_list = replicate_filter
else:
process_scalar = reflect
process_list = reflect_filter
value = getattr(source, attr.key)
target_attr_model = target_attr.property.mapper.class_
if attr.property.uselist:
adapter = collection_adapter(value)
if adapter:
# Convert any collection to flat iterable
value = adapter.adapt_like_to_iterable(value)
reflection = process_list(value, target_attr_model, cache=cache)
impl = instance_state(target).get_impl(attr.key)
# Set any collection value from flat list
impl._set_iterable(instance_state(target),
instance_dict(target),
reflection)
else:
reflection = process_scalar(value, target_attr_model, cache=cache)
setattr(target, attr.key, reflection)
if (reflection is None and
attr.property.direction is MANYTOONE and
any(col.primary_key and not col.nullable
for col in attr.property.local_columns)):
raise _PrimaryKeyIsNull()
开发者ID:Lehych,项目名称:iktomi,代码行数:28,代码来源:replication.py
示例4: test_basic
def test_basic(self):
for base in (object, MyBaseClass, MyClass):
class User(base):
pass
register_class(User)
attributes.register_attribute(
User, 'user_id', uselist=False, useobject=False)
attributes.register_attribute(
User, 'user_name', uselist=False, useobject=False)
attributes.register_attribute(
User, 'email_address', uselist=False, useobject=False)
u = User()
u.user_id = 7
u.user_name = 'john'
u.email_address = '[email protected]'
eq_(u.user_id, 7)
eq_(u.user_name, "john")
eq_(u.email_address, "[email protected]")
attributes.instance_state(u)._commit_all(
attributes.instance_dict(u))
eq_(u.user_id, 7)
eq_(u.user_name, "john")
eq_(u.email_address, "[email protected]")
u.user_name = 'heythere'
u.email_address = '[email protected]'
eq_(u.user_id, 7)
eq_(u.user_name, "heythere")
eq_(u.email_address, "[email protected]")
开发者ID:anti-social,项目名称:sqlalchemy,代码行数:32,代码来源:test_extendedattr.py
示例5: replicate_relation
def replicate_relation(source, target, attr, target_attr, cache=None):
if attr.property.cascade.delete_orphan:
process_scalar = replicate_no_merge
process_list = replicate_filter
else:
process_scalar = reflect
process_list = reflect_filter
value = getattr(source, attr.key)
target_attr_model = target_attr.property.mapper.class_
if attr.property.uselist:
adapter = collection_adapter(value)
if adapter:
# XXX The magic passes below are adapted from logic in
# CollectionAttributeImpl.set() method without proper
# understanding. The `elif` branch isn't even coverered by tests.
if hasattr(value, '_sa_iterator'):
value = value._sa_iterator()
elif duck_type_collection(value) is dict:
value = value.values()
reflection = process_list(value, target_attr_model, cache=cache)
impl = instance_state(target).get_impl(attr.key)
impl.set(instance_state(target), instance_dict(target), reflection,
# XXX We either have to convert reflection back to original
# collection type or use this private parameter.
_adapt=False)
else:
reflection = process_scalar(value, target_attr_model, cache=cache)
setattr(target, attr.key, reflection)
if (reflection is None and
attr.property.direction is MANYTOONE and
any(col.primary_key and not col.nullable
for col in attr.property.local_columns)):
raise _PrimaryKeyIsNull()
开发者ID:SmartTeleMax,项目名称:iktomi,代码行数:33,代码来源:replication.py
示例6: test_history_populated_passive_return_never_set
def test_history_populated_passive_return_never_set(self):
User, Address, sess, a1 = self._u_ad_fixture(True)
eq_(
Address.user.impl.get_history(
attributes.instance_state(a1), attributes.instance_dict(a1), passive=attributes.PASSIVE_RETURN_NEVER_SET
),
((), [User(name="ed")], ()),
)
开发者ID:vishvananda,项目名称:sqlalchemy,代码行数:8,代码来源:test_lazy_relations.py
示例7: fdel
def fdel(instance):
state = attributes.instance_state(instance)
dict_ = attributes.instance_dict(instance)
previous = dict_.pop(self.key, attributes.NO_VALUE)
attr = state.manager[self.key]
attr.dispatch.remove(state, previous, attr.impl)
for key in self._attribute_keys:
setattr(instance, key, None)
开发者ID:MorganBorman,项目名称:cxsbs,代码行数:8,代码来源:descriptor_props.py
示例8: cache_list_version_key
def cache_list_version_key(self, data=None):
if not self._meta.cache_list_keys:
raise Exception('Class._meta has no cache_list_keys')
if not has_identity(self):
raise Exception('Cannot generate list cache key for instance ' \
'with no identity')
data = data or instance_dict(self)
raw_key, attrs = self._meta.cache_list_keys[0]
return raw_key % data
开发者ID:dieselmachine,项目名称:baph,代码行数:9,代码来源:mixins.py
示例9: cache_detail_key
def cache_detail_key(self, data=None):
if not hasattr(self._meta, 'cache_detail_keys'):
raise Exception('Class meta has no cache_detail_keys')
if not has_identity(self):
raise Exception('Cannot generate detail cache key for instance ' \
'with no identity')
data = data or instance_dict(self)
raw_key, attrs = self._meta.cache_detail_keys[0]
return self.format_key(raw_key % data)
开发者ID:dieselmachine,项目名称:baph,代码行数:9,代码来源:mixins.py
示例10: cache_pointers
def cache_pointers(self, data=None, columns=[]):
if not hasattr(self._meta, 'cache_pointers'):
return {}
data = data or instance_dict(self)
keys = {}
for raw_key, attrs, name in self._meta.cache_pointers:
if columns and not any(c in attrs for c in columns):
continue
keys[name] = raw_key % data
return keys
开发者ID:devhub,项目名称:baph,代码行数:10,代码来源:mixins.py
示例11: test_history_empty_passive_return_never_set
def test_history_empty_passive_return_never_set(self):
User, Address, sess, a1 = self._u_ad_fixture(False)
eq_(
Address.user.impl.get_history(
attributes.instance_state(a1), attributes.instance_dict(a1), passive=attributes.PASSIVE_RETURN_NEVER_SET
),
((), (), ()),
)
assert "user_id" not in a1.__dict__
assert "user" not in a1.__dict__
开发者ID:vishvananda,项目名称:sqlalchemy,代码行数:10,代码来源:test_lazy_relations.py
示例12: test_get_empty_passive_no_initialize
def test_get_empty_passive_no_initialize(self):
User, Address, sess, a1 = self._u_ad_fixture(False)
eq_(
Address.user.impl.get(
attributes.instance_state(a1), attributes.instance_dict(a1), passive=attributes.PASSIVE_NO_INITIALIZE
),
attributes.PASSIVE_NO_RESULT,
)
assert "user_id" not in a1.__dict__
assert "user" not in a1.__dict__
开发者ID:vishvananda,项目名称:sqlalchemy,代码行数:10,代码来源:test_lazy_relations.py
示例13: test_history_populated_passive_no_initialize
def test_history_populated_passive_no_initialize(self):
User, Address, sess, a1 = self._u_ad_fixture(True)
eq_(
Address.user.impl.get_history(
attributes.instance_state(a1), attributes.instance_dict(a1), passive=attributes.PASSIVE_NO_INITIALIZE
),
attributes.HISTORY_BLANK,
)
assert "user_id" not in a1.__dict__
assert "user" not in a1.__dict__
开发者ID:vishvananda,项目名称:sqlalchemy,代码行数:10,代码来源:test_lazy_relations.py
示例14: test_get_populated_passive_return_never_set
def test_get_populated_passive_return_never_set(self):
User, Address, sess, a1 = self._u_ad_fixture(True)
eq_(
Address.user.impl.get(
attributes.instance_state(a1),
attributes.instance_dict(a1),
passive=attributes.PASSIVE_RETURN_NO_VALUE,
),
User(name="ed"),
)
开发者ID:monetate,项目名称:sqlalchemy,代码行数:10,代码来源:test_lazy_relations.py
示例15: lazy_clause
def lazy_clause(self, state, reverse_direction=False,
alias_secondary=False,
adapt_source=None):
if state is None:
return self._lazy_none_clause(
reverse_direction,
adapt_source=adapt_source)
if not reverse_direction:
criterion, bind_to_col, rev = \
self.__lazywhere, \
self.__bind_to_col, \
self._equated_columns
else:
criterion, bind_to_col, rev = \
LazyLoader._create_lazy_clause(
self.parent_property,
reverse_direction=reverse_direction)
if reverse_direction:
mapper = self.parent_property.mapper
else:
mapper = self.parent_property.parent
o = state.obj() # strong ref
dict_ = attributes.instance_dict(o)
# use the "committed state" only if we're in a flush
# for this state.
sess = sessionlib._state_session(state)
if sess is not None and sess._flushing:
def visit_bindparam(bindparam):
if bindparam.key in bind_to_col:
bindparam.callable = \
lambda: mapper._get_committed_state_attr_by_column(
state, dict_, bind_to_col[bindparam.key])
else:
def visit_bindparam(bindparam):
if bindparam.key in bind_to_col:
bindparam.callable = lambda: mapper._get_state_attr_by_column(
state, dict_, bind_to_col[bindparam.key])
if self.parent_property.secondary is not None and alias_secondary:
criterion = sql_util.ClauseAdapter(
self.parent_property.secondary.alias()).\
traverse(criterion)
criterion = visitors.cloned_traverse(
criterion, {}, {'bindparam':visit_bindparam})
if adapt_source:
criterion = adapt_source(criterion)
return criterion
开发者ID:denny820909,项目名称:builder,代码行数:55,代码来源:strategies.py
示例16: test_get_populated_passive_no_initialize
def test_get_populated_passive_no_initialize(self):
User, Address, sess, a1 = self._u_ad_fixture(True)
eq_(
Address.user.impl.get(
attributes.instance_state(a1),
attributes.instance_dict(a1),
passive=attributes.PASSIVE_NO_INITIALIZE),
attributes.PASSIVE_NO_RESULT
)
assert 'user_id' not in a1.__dict__
assert 'user' not in a1.__dict__
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:11,代码来源:test_lazy_relations.py
示例17: blacklist
def blacklist(self, limit = None):
"""Return blacklisted items in the database, with optional limit."""
if (limit is not None):
rows = self.session.query(Messages).filter(Messages.message_blacklist == True).order_by(desc(Messages.message_priority)).order_by(desc(Messages.message_received_timestamp)).all()[0:limit]
else:
rows = self.session.query(Messages).filter(Messages.message_blacklist == True).order_by(desc(Messages.message_priority)).order_by(desc(Messages.message_received_timestamp)).all()
results = []
for row in rows:
results.append(instance_dict(row))
return results
开发者ID:zeitkunst,项目名称:FluidNexus,代码行数:12,代码来源:Database.py
示例18: test_history
def test_history(self):
for base in (object, MyBaseClass, MyClass):
class Foo(base):
pass
class Bar(base):
pass
attributes.register_class(Foo)
attributes.register_class(Bar)
attributes.register_attribute(Foo, "name", uselist=False, useobject=False)
attributes.register_attribute(Foo, "bars", uselist=True, trackparent=True, useobject=True)
attributes.register_attribute(Bar, "name", uselist=False, useobject=False)
f1 = Foo()
f1.name = 'f1'
eq_(attributes.get_state_history(attributes.instance_state(f1), 'name'), (['f1'], (), ()))
b1 = Bar()
b1.name = 'b1'
f1.bars.append(b1)
eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ([b1], [], []))
attributes.instance_state(f1).commit_all(attributes.instance_dict(f1))
attributes.instance_state(b1).commit_all(attributes.instance_dict(b1))
eq_(attributes.get_state_history(attributes.instance_state(f1), 'name'), ((), ['f1'], ()))
eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ((), [b1], ()))
f1.name = 'f1mod'
b2 = Bar()
b2.name = 'b2'
f1.bars.append(b2)
eq_(attributes.get_state_history(attributes.instance_state(f1), 'name'), (['f1mod'], (), ['f1']))
eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ([b2], [b1], []))
f1.bars.remove(b1)
eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ([b2], [], [b1]))
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:38,代码来源:test_extendedattr.py
示例19: fset
def fset(instance, value):
dict_ = attributes.instance_dict(instance)
state = attributes.instance_state(instance)
attr = state.manager[self.key]
previous = dict_.get(self.key, attributes.NO_VALUE)
for fn in attr.dispatch.set:
value = fn(state, value, previous, attr.impl)
dict_[self.key] = value
if value is None:
for key in self._attribute_keys:
setattr(instance, key, None)
else:
for key, value in zip(
self._attribute_keys,
value.__composite_values__()):
setattr(instance, key, value)
开发者ID:MorganBorman,项目名称:cxsbs,代码行数:16,代码来源:descriptor_props.py
示例20: fget
def fget(instance):
dict_ = attributes.instance_dict(instance)
if self.key not in dict_:
# key not present. Iterate through related
# attributes, retrieve their values. This
# ensures they all load.
values = [getattr(instance, key) for key in self._attribute_keys]
# usually, the load() event will have loaded our key
# at this point, unless we only loaded relationship()
# attributes above. Populate here if that's the case.
if self.key not in dict_ and not _none_set.issuperset(values):
dict_[self.key] = self.composite_class(*values)
return dict_.get(self.key, None)
开发者ID:MorganBorman,项目名称:cxsbs,代码行数:16,代码来源:descriptor_props.py
注:本文中的sqlalchemy.orm.attributes.instance_dict函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论