1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
"""
Illustrates an extension which creates version tables for entities and stores
records for each change. The given extensions generate an anonymous "history"
class which represents historical versions of the target object.
Compare to the :ref:`examples_versioned_rows` examples which write updates
as new rows in the same table, without using a separate history table.
Usage is illustrated via a unit test module ``test_versioning.py``, which is
run using SQLAlchemy's internal pytest plugin::
$ pytest test/base/test_examples.py
A fragment of example usage, using declarative::
from history_meta import Versioned, versioned_session
class Base(DeclarativeBase):
pass
class SomeClass(Versioned, Base):
__tablename__ = "sometable"
id = Column(Integer, primary_key=True)
name = Column(String(50))
def __eq__(self, other):
assert type(other) is SomeClass and other.id == self.id
Session = sessionmaker(bind=engine)
versioned_session(Session)
sess = Session()
sc = SomeClass(name="sc1")
sess.add(sc)
sess.commit()
sc.name = "sc1modified"
sess.commit()
assert sc.version == 2
SomeClassHistory = SomeClass.__history_mapper__.class_
assert sess.query(SomeClassHistory).filter(
SomeClassHistory.version == 1
).all() == [SomeClassHistory(version=1, name="sc1")]
The ``Versioned`` mixin is designed to work with declarative. To use
the extension with classical mappers, the ``_history_mapper`` function
can be applied::
from history_meta import _history_mapper
m = mapper(SomeClass, sometable)
_history_mapper(m)
SomeHistoryClass = SomeClass.__history_mapper__.class_
The versioning example also integrates with the ORM optimistic concurrency
feature documented at :ref:`mapper_version_counter`. To enable this feature,
set the flag ``Versioned.use_mapper_versioning`` to True::
class SomeClass(Versioned, Base):
__tablename__ = "sometable"
use_mapper_versioning = True
id = Column(Integer, primary_key=True)
name = Column(String(50))
def __eq__(self, other):
assert type(other) is SomeClass and other.id == self.id
Above, if two instance of ``SomeClass`` with the same version identifier
are updated and sent to the database for UPDATE concurrently, if the database
isolation level allows the two UPDATE statements to proceed, one will fail
because it no longer is against the last known version identifier.
.. autosource::
"""
|
