SQLAchemy scoped_session问题
SQLAchemy scoped_session问题
提问于 2022-11-16 15:02:59
经过很长一段时间的工作，我仍然有一些关于sqlalchemy作用域会话的问题，我无法弄清楚。例如，我为提供session的函数设置了装饰器。
def db_session_provider(commit=True, rollback=True, reraise=True):
def decorator(func: typing.Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
with Session() as session:
try:
result = func(*args, **kwargs, session=session)
if commit:
session.commit()
return result
except: # noqa
if rollback:
session.rollback()
if reraise:
raise
return wrapper
return decorator其中Session是构造器，定义如下：
session_factory = sessionmaker(
autocommit=config.SQLALCHEMY_AUTOCOMMIT, autoflush=config.SQLALCHEMY_AUTOFLUSH, bind=engine, expire_on_commit=False
)
Session = scoped_session(sessionmaker())现在，我有错误sqlalchemy.orm.exc.DetachedInstanceError: Instance <Client at 0x10daae430> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/14/bhk3)失败的代码。链接文档并不能使事情变得更清楚，因为看起来不相关。
下面是触发此类错误的代码：
def fn_with_ext_session(client: Client, session: Session) -> None:
# do something with client, it is legit and works
print(f"Client {client.id} fetched")
@db_session_provider()
def fn_with_int_session(client_id: int, session: Session) -> None:
# doing stuff unrelated to model Client but involves some other linked tables:
# here `session` passed by decorator
trades = session.query(Trade).filter(Trade.client_id == client_id).all()
# after exiting from this function outer object `Client` becomes detached!
@db_session_provider()
def fn1(session: Session):
client = session.query(Client).get(1)
# here Client attached to the session
fn_with_ext_session(client, session)
# here Client attached to the session
fn_with_int_session(client.id)
# here Client DETACHED from locally defined session!!!
print(f"Client {client.id}") # <--- here exception raised请你澄清一下炼金术会议是如何生活的，为什么它在这里重叠？
回答 1
Stack Overflow用户
发布于 2022-11-16 23:26:45
Base = declarative_base()
engine = create_engine(f"postgresql+psycopg2://{username}:{password}@/{db}", echo=False)
class Client(Base):
__tablename__ = "clients"
id = Column(
Integer, nullable=False, primary_key=True
)
name = Column(Text)
Base.metadata.create_all(engine)
session_factory = sessionmaker(
autocommit=False, autoflush=False, bind=engine, expire_on_commit=False
)
Session = scoped_session(session_factory)
def db_session_provider(commit=True, rollback=True, reraise=True):
def decorator(func: typing.Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
with Session() as session:
try:
result = func(*args, **kwargs, session=session)
if commit:
session.commit()
return result
except: # noqa
if rollback:
session.rollback()
if reraise:
raise
return wrapper
return decorator
@db_session_provider()
def create_clients(session: Session):
c1 = Client(name="first client")
session.add(c1)
c2 = Client(name="second client")
session.add(c2)
@db_session_provider()
def read_clients(session: Session):
print ([c.name for c in session.query(Client).all()])
create_clients()
read_clients()输出
['first client', 'second client']页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/74462772
复制相关文章
相似问题
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号