this is a simple example of what youre trying to do. its not even as
fancy as what i described above and just uses a threadlocal to keep
track of the current Session:
from sqlalchemy import *
from sqlalchemy.orm.session import object_session
from cStringIO import StringIO
from pickle import Pickler, Unpickler
import threading
meta = BoundMetaData('sqlite://', echo=True)
class MyExt(MapperExtension):
def populate_instance(self, mapper, selectcontext, row, instance,
identitykey, isnew):
# if you access the row before this step, it fails
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 hasattr(obj, "id"):
key = "%s:%s" % (type(obj).__name__, obj.id)
return key
return None
def persistent_load(self, key):
print "KEY", 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()
b = Bar('some bar')
sess.save(b)
sess.flush()
sess.clear()
sess = create_session()
f = Foo()
f.bar = b
sess.save(f)
sess.flush()
sess.clear()
# delete current session to make sure its populated
del MyPickler.sessions.current
f = sess.query(Foo).get(f.id)
assert f.bar.data == 'some bar'
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
