本文整理汇总了Python中sqlalchemy.orm.attributes.get_history函数的典型用法代码示例。如果您正苦于以下问题:Python get_history函数的具体用法?Python get_history怎么用?Python get_history使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_history函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: get_attribute_history
def get_attribute_history(self, state, key, passive=True):
hashkey = ("history", state, key)
# cache the objects, not the states; the strong reference here
# prevents newly loaded objects from being dereferenced during the
# flush process
if hashkey in self.attributes:
(added, unchanged, deleted, cached_passive) = self.attributes[hashkey]
# if the cached lookup was "passive" and now we want non-passive, do a non-passive
# lookup and re-cache
if cached_passive and not passive:
(added, unchanged, deleted) = attributes.get_history(state, key, passive=False)
self.attributes[hashkey] = (added, unchanged, deleted, passive)
else:
(added, unchanged, deleted) = attributes.get_history(state, key, passive=passive)
self.attributes[hashkey] = (added, unchanged, deleted, passive)
if added is None:
return (added, unchanged, deleted)
else:
return (
[getattr(c, '_state', c) for c in added],
[getattr(c, '_state', c) for c in unchanged],
[getattr(c, '_state', c) for c in deleted],
)
开发者ID:Frihet,项目名称:sqlalchemy-patches,代码行数:25,代码来源:unitofwork.py
示例2: test_lazy_backref_collections
def test_lazy_backref_collections(self):
class Foo(_base.BasicEntity):
pass
class Bar(_base.BasicEntity):
pass
lazy_load = []
def lazyload(instance):
def load():
return lazy_load
return load
attributes.register_class(Foo)
attributes.register_class(Bar)
attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, callable_=lazyload, useobject=True)
attributes.register_attribute(Bar, 'foo', uselist=False, extension=attributes.GenericBackrefExtension('bars'), trackparent=True, useobject=True)
bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
lazy_load = [bar1, bar2, bar3]
f = Foo()
bar4 = Bar()
bar4.foo = f
eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []))
lazy_load = None
f = Foo()
bar4 = Bar()
bar4.foo = f
eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [], []))
lazy_load = [bar1, bar2, bar3]
attributes.instance_state(f).expire_attributes(['bars'])
eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ((), [bar1, bar2, bar3], ()))
开发者ID:jrus,项目名称:sqlalchemy,代码行数:34,代码来源:attributes.py
示例3: test_many_to_one_cascade
def test_many_to_one_cascade(self):
mapper(Address, addresses, properties={
'user':relationship(User)
})
mapper(User, users)
u1 = User(id=1, name="u1")
a1 =Address(id=1, email_address="a1", user=u1)
u2 = User(id=2, name="u2")
sess = create_session()
sess.add_all([a1, u2])
sess.flush()
a1.user = u2
sess2 = create_session()
a2 = sess2.merge(a1)
eq_(
attributes.get_history(a2, 'user'),
([u2], (), [attributes.PASSIVE_NO_RESULT])
)
assert a2 in sess2.dirty
sess.refresh(a1)
sess2 = create_session()
a2 = sess2.merge(a1, load=False)
eq_(
attributes.get_history(a2, 'user'),
((), [u1], ())
)
assert a2 not in sess2.dirty
开发者ID:chatch,项目名称:pinyin-toolkit,代码行数:33,代码来源:test_merge.py
示例4: edit_task
def edit_task(name, goal, strategy, task):
project = models.Projects.query.filter_by(id=name).first()
pgoal = models.Goals.query.filter_by(id=goal).first()
pstrat = models.Strategies.query.filter_by(id=strategy).first()
ptask = models.Tasks.query.filter_by(id=task).first()
form = task_form(obj=ptask)
form.populate_obj(ptask)
form.deadline.data = ptask.deadline.strftime("%m/%d/%Y")
tform = task_form(request.values)
if request.method == "POST" and form.validate_on_submit():
# if it changed from True to false, set complete date to None
if get_history(ptask, "complete")[0] == [True] and get_history(ptask, "complete")[2] == [False]:
print "changed from false to true"
ptask.completeDate = datetime.datetime.utcnow()
if get_history(ptask, "complete")[0] == [False] and get_history(ptask, "complete")[2] == [True]:
print "changed from true to false"
ptask.completeDate = None
# task=tform.task.data
# strat=pstrat
# note = tform.note.data
# staff=tform.staff.data
# deadline=tform.deadline.data
# complete=tform.complete.data
# created=datetime.datetime.utcnow()
db.session.commit()
return redirect(url_for("task_outline", name=name, goal=goal, strategy=strategy))
return render_template(
"edit_task.html", tform=tform, form=form, project=project, pgoal=pgoal, pstrat=pstrat, ptask=ptask
)
开发者ID:chetstar,项目名称:projectmgmt-and-restart,代码行数:29,代码来源:viewsort.py
示例5: test_dict_collections
def test_dict_collections(self):
class Foo(fixtures.Base):
pass
class Bar(fixtures.Base):
pass
from sqlalchemy.orm.collections import attribute_mapped_collection
attributes.register_class(Foo)
attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))
hi = Bar(name='hi')
there = Bar(name='there')
old = Bar(name='old')
new = Bar(name='new')
f = Foo()
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], []))
f.someattr['hi'] = hi
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], []))
f.someattr['there'] = there
self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([hi, there]), set([]), set([])))
f._state.commit(['someattr'])
self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([]), set([hi, there]), set([])))
开发者ID:Frihet,项目名称:sqlalchemy-patches,代码行数:27,代码来源:attributes.py
示例6: test_dict_collections
def test_dict_collections(self):
class Foo(_base.BasicEntity):
pass
class Bar(_base.BasicEntity):
pass
from sqlalchemy.orm.collections import attribute_mapped_collection
attributes.register_class(Foo)
attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))
hi = Bar(name='hi')
there = Bar(name='there')
old = Bar(name='old')
new = Bar(name='new')
f = Foo()
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [], ()))
f.someattr['hi'] = hi
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))
f.someattr['there'] = there
eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set([hi, there]), set(), set()))
attributes.instance_state(f).commit(['someattr'])
eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set(), set([hi, there]), set()))
开发者ID:jrus,项目名称:sqlalchemy,代码行数:27,代码来源:attributes.py
示例7: channel_visibility_change
def channel_visibility_change(mapper, connection, target):
if (
get_history(target, "public").has_changes()
or get_history(target, "deleted").has_changes()
or get_history(target, "visible").has_changes()
):
instances = list(VideoInstance.query.filter(VideoInstance.channel == target.id).values("id"))
if instances:
update_video_instance_date_updated([i[0] for i in instances], visible=_channel_is_public(target))
开发者ID:wonderpl,项目名称:dolly-web,代码行数:10,代码来源:models.py
示例8: leasing_force_expiration
def leasing_force_expiration(mapper, connection, target):
expired = False
added, unchanged, deleted = get_history(target, 'static_ip')
expired = expired or added or deleted
added, unchanged, deleted = get_history(target, 'pool_subnet')
expired = expired or added or deleted
if expired:
Lease.query.with_parent(target).update({'force_expire': True})
开发者ID:andreydomas,项目名称:pybootstrapper,代码行数:12,代码来源:models.py
示例9: _assert_history
def _assert_history(self, obj, compare, compare_passive=None):
if isinstance(obj, self.classes.User):
attrname = "addresses"
elif isinstance(obj, self.classes.Order):
attrname = "items"
eq_(attributes.get_history(obj, attrname), compare)
if compare_passive is None:
compare_passive = compare
eq_(attributes.get_history(obj, attrname, attributes.LOAD_AGAINST_COMMITTED), compare_passive)
开发者ID:hunterfu,项目名称:LuoYunCloud,代码行数:12,代码来源:test_dynamic.py
示例10: test_get_history
def test_get_history(self):
Edge = self.classes.Edge
Point = self.classes.Point
from sqlalchemy.orm.attributes import get_history
e1 = Edge()
e1.start = Point(1, 2)
eq_(
get_history(e1, "start"),
([Point(x=1, y=2)], (), [Point(x=None, y=None)]),
)
eq_(get_history(e1, "end"), ((), [Point(x=None, y=None)], ()))
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:13,代码来源:test_composites.py
示例11: test_object_collections_set
def test_object_collections_set(self):
class Foo(_base.BasicEntity):
pass
class Bar(_base.BasicEntity):
def __nonzero__(self):
assert False
attributes.register_class(Foo)
attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)
hi = Bar(name='hi')
there = Bar(name='there')
old = Bar(name='old')
new = Bar(name='new')
# case 1. new object
f = Foo()
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [], ()))
f.someattr = [hi]
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))
attributes.instance_state(f).commit(['someattr'])
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [hi], ()))
f.someattr = [there]
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [], [hi]))
attributes.instance_state(f).commit(['someattr'])
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [there], ()))
f.someattr = [hi]
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [there]))
f.someattr = [old, new]
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old, new], [], [there]))
# case 2. object with direct settings (similar to a load operation)
f = Foo()
collection = attributes.init_collection(attributes.instance_state(f), 'someattr')
collection.append_without_event(new)
attributes.instance_state(f).commit_all()
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [new], ()))
f.someattr = [old]
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old], [], [new]))
attributes.instance_state(f).commit(['someattr'])
eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [old], ()))
开发者ID:jrus,项目名称:sqlalchemy,代码行数:50,代码来源:attributes.py
示例12: test_object_collections_set
def test_object_collections_set(self):
class Foo(fixtures.Base):
pass
class Bar(fixtures.Base):
def __nonzero__(self):
assert False
attributes.register_class(Foo)
attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)
hi = Bar(name='hi')
there = Bar(name='there')
old = Bar(name='old')
new = Bar(name='new')
# case 1. new object
f = Foo()
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], []))
f.someattr = [hi]
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], []))
f._state.commit(['someattr'])
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], []))
f.someattr = [there]
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [hi]))
f._state.commit(['someattr'])
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [there], []))
f.someattr = [hi]
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [there]))
f.someattr = [old, new]
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old, new], [], [there]))
# case 2. object with direct settings (similar to a load operation)
f = Foo()
collection = attributes.init_collection(f, 'someattr')
collection.append_without_event(new)
f._state.commit_all()
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], []))
f.someattr = [old]
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [], [new]))
f._state.commit(['someattr'])
self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old], []))
开发者ID:Frihet,项目名称:sqlalchemy-patches,代码行数:50,代码来源:attributes.py
示例13: test_lazyhistory
def test_lazyhistory(self):
"""tests that history functions work with lazy-loading attributes"""
class Foo(_base.BasicEntity):
pass
class Bar(_base.BasicEntity):
pass
attributes.register_class(Foo)
attributes.register_class(Bar)
bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
def func1():
return "this is func 1"
def func2():
return [bar1, bar2, bar3]
attributes.register_attribute(Foo, 'col1', uselist=False, callable_=lambda o:func1, useobject=True)
attributes.register_attribute(Foo, 'col2', uselist=True, callable_=lambda o:func2, useobject=True)
attributes.register_attribute(Bar, 'id', uselist=False, useobject=True)
x = Foo()
attributes.instance_state(x).commit_all()
x.col2.append(bar4)
eq_(attributes.get_history(attributes.instance_state(x), 'col2'), ([bar4], [bar1, bar2, bar3], []))
开发者ID:jrus,项目名称:sqlalchemy,代码行数:25,代码来源:attributes.py
示例14: on_model_change
def on_model_change(self, form, model, is_created):
if form.password.data is None or len(form.password.data) < 3:
prev_hash = get_history(model, 'password_hash')[2][0]
model.password_hash = prev_hash
db.session.commit()
pass
开发者ID:greginvm,项目名称:farmcontrol,代码行数:7,代码来源:views.py
示例15: after_update
def after_update(self):
client_id_hist = get_history(self.client_id)
if client_id_hist.deleted:
session = orm.Session.object_session(self)
session.pipe.delete(self._cache_name_client_id.format(
client_id_hist.deleted[0]
))
开发者ID:thomaserlang,项目名称:seplis,代码行数:7,代码来源:app.py
示例16: 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
示例17: _record
def _record(self, mapper, model, operation):
pk = tuple(mapper.primary_key_from_instance(model))
#orm.object_session(model)._model_changes[pk] = (model, operation)
changes = {}
for prop in object_mapper(model).iterate_properties:
if not isinstance(prop, RelationshipProperty):
try:
history = attributes.get_history(model, prop.key)
except:
continue
added, unchanged, deleted = history
newvalue = added[0] if added else None
if operation=='delete':
oldvalue = unchanged[0] if unchanged else None
else:
oldvalue = deleted[0] if deleted else None
if newvalue or oldvalue:
changes[prop.key] = (oldvalue, newvalue)
orm.object_session(model)._model_changes[pk] = (model.__tablename__, pk[0], changes, operation)
return EXT_CONTINUE
开发者ID:Magnil,项目名称:pypress-tornado,代码行数:26,代码来源:sqlalchemy.py
示例18: attrs_changed
def attrs_changed(obj, *attrs):
"""Checks if the given fields have been changed since the last flush
:param obj: SQLAlchemy-mapped object
:param attrs: attribute names
"""
return any(get_history(obj, attr).has_changes() for attr in attrs)
开发者ID:dbourillot,项目名称:indico,代码行数:7,代码来源:models.py
示例19: create_version
def create_version(obj, session, deleted = False):
obj_mapper = object_mapper(obj)
history_mapper = obj.__history_mapper__
history_cls = history_mapper.class_
obj_state = attributes.instance_state(obj)
attr = {}
obj_changed = False
for om, hm in zip(obj_mapper.iterate_to_root(), history_mapper.iterate_to_root()):
if hm.single:
continue
for hist_col in hm.local_table.c:
if hist_col.key == 'version':
continue
obj_col = om.local_table.c[hist_col.key]
# get the value of the
# attribute based on the MapperProperty related to the
# mapped column. this will allow usage of MapperProperties
# that have a different keyname than that of the mapped column.
try:
prop = obj_mapper.get_property_by_column(obj_col)
except UnmappedColumnError:
# in the case of single table inheritance, there may be
# columns on the mapped table intended for the subclass only.
# the "unmapped" status of the subclass column on the
# base class is a feature of the declarative module as of sqla 0.5.2.
continue
# expired object attributes and also deferred cols might not be in the
# dict. force it to load no matter what by using getattr().
if prop.key not in obj_state.dict:
getattr(obj, prop.key)
a, u, d = attributes.get_history(obj, prop.key)
if d:
attr[hist_col.key] = d[0]
obj_changed = True
elif u:
attr[hist_col.key] = u[0]
else:
# if the attribute had no value.
attr[hist_col.key] = a[0]
obj_changed = True
if not obj_changed and not deleted:
return
attr['version'] = obj.version
hist = history_cls()
for key, value in attr.iteritems():
setattr(hist, key, value)
session.add(hist)
obj.version += 1
开发者ID:AndryulE,项目名称:kitsune,代码行数:60,代码来源:history_meta.py
示例20: changeset
def changeset(obj):
"""
Returns a humanized changeset for given SQLAlchemy declarative object.
:param obj: SQLAlchemy declarative model object
"""
data = {}
session = sa.orm.object_session(obj)
if session and obj in session.deleted:
for prop in obj.__mapper__.iterate_properties:
if isinstance(prop, sa.orm.ColumnProperty):
if not prop.columns[0].primary_key:
value = getattr(obj, prop.key)
if value is not None:
data[prop.key] = [None, getattr(obj, prop.key)]
else:
for prop in obj.__mapper__.iterate_properties:
history = get_history(obj, prop.key)
if history.has_changes():
old_value = history.deleted[0] if history.deleted else None
new_value = history.added[0] if history.added else None
if new_value:
data[prop.key] = [new_value, old_value]
return data
开发者ID:FelixLoether,项目名称:sqlalchemy-continuum,代码行数:25,代码来源:utils.py
注:本文中的sqlalchemy.orm.attributes.get_history函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论