How do I effectively use SQLAlchemy with multiple DDD Repositories?

Question

I have been trying to find some examples of how to implement the Repository pattern with SQLAlchemy. Specifically, implementing more than one Repository.

In the case of multiple Repositories, I believe each Repository would be best implemented by maintaining a separate SQLAlchemy session. However, I have been running into a problem trying to move an instance of an object bound to one session to another session.

First, does this make sense to do? Should each Repository maintain its own UoW separate from any other Repository or should it be considered safe to have the entire context share the same Session?

Second, what is the best way to detach an instance from one Session and bind it to another?

Third, are there any solid DDD Repository examples written with SQLAlchemy in mind?

Answer 1

I'm not familar with DDD Repository pattern, but below is an exmaple showing how to move an object from one session to another:

from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

metadata  = MetaData()
Base = declarative_base(metadata=metadata, name='Base')

class Model(Base):
    __tablename__ = 'models'
    id = Column(Integer, primary_key=True)


engine1 = create_engine('sqlite://')
metadata.create_all(engine1)
engine2 = create_engine('sqlite://')
metadata.create_all(engine2)

session1 = sessionmaker(bind=engine1)()
session2 = sessionmaker(bind=engine2)()

# Setup an single object in the first repo.
obj = Model()
session1.add(obj)
session1.commit()
session1.expunge_all() 

# Move object from the first repo to the second.
obj = session1.query(Model).first()
assert session2.query(Model).count()==0
session1.delete(obj)
# You have to flush before expunging, otherwise it won't be deleted.
session1.flush()
session1.expunge(obj)
obj = session2.merge(obj)
# An optimistic way to bind two transactions is flushing before commiting.
session2.flush()
session1.commit()
session2.commit()
assert session1.query(Model).count()==0
assert session2.query(Model).count()==1