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
|
"""illustrates one way to use a custom pickler that is session-aware."""
from sqlalchemy import *
from sqlalchemy.orm.session import object_session
from cStringIO import StringIO
from pickle import Pickler, Unpickler
import threading
meta = MetaData('sqlite://')
meta.bind.echo = True
class MyExt(MapperExtension):
def populate_instance(self, mapper, selectcontext, row, instance, identitykey, isnew):
MyPickler.sessions.current = selectcontext.session
return EXT_PASS
def before_insert(self, mapper, connection, instance):
MyPickler.sessions.current = object_session(instance)
return EXT_PASS
def before_update(self, mapper, connection, instance):
MyPickler.sessions.current = object_session(instance)
return EXT_PASS
class MyPickler(object):
sessions = threading.local()
def persistent_id(self, obj):
if getattr(obj, "id", None) is None:
sess = MyPickler.sessions.current
newsess = create_session(bind=sess.connection(class_mapper(Bar)))
newsess.save(obj)
newsess.flush()
key = "%s:%s" % (type(obj).__name__, obj.id)
return key
def persistent_load(self, key):
name, ident = key.split(":")
sess = MyPickler.sessions.current
return sess.query(Bar).get(ident)
def dumps(self, graph, protocol):
src = StringIO()
pickler = Pickler(src)
pickler.persistent_id = self.persistent_id
pickler.dump(graph)
return src.getvalue()
def loads(self, data):
dst = StringIO(data)
unpickler = Unpickler(dst)
unpickler.persistent_load = self.persistent_load
return unpickler.load()
foo_table = Table('foo', meta,
Column('id', Integer, primary_key=True),
Column('bar', PickleType(pickler=MyPickler()), nullable=False))
bar_table = Table('bar', meta,
Column('id', Integer, primary_key=True),
Column('data', String(40)))
meta.create_all()
class Foo(object):
pass
class Bar(object):
def __init__(self, value):
self.data = value
mapper(Foo, foo_table, extension=MyExt())
mapper(Bar, bar_table)
sess = create_session()
f = Foo()
f.bar = Bar('some bar')
sess.save(f)
sess.flush()
sess.clear()
del MyPickler.sessions.current
f = sess.query(Foo).get(f.id)
assert f.bar.data == 'some bar'
|
