在给定项目结构中，在烧瓶应用程序中管理 sqlalchemy 会话的最佳方法是什么?

Question

嗨，对不起，这是一个非常开放的问题，更像是一个编码建议。寻找在烧瓶项目中使用 sqllchemy 会话的最佳实践是什么。找不到可以回答我问题的博客，因此在这里问。

我有一个烧瓶应用程序的项目结构，类似于以下内容：

---services
--- service_1.py
--- service_2.py
---models.py
---adapter.py
---app.py

所有业务逻辑都驻留在服务中。服务中的方法调用适配器中的方法，以便与数据库进行任何交互。

适配器是数据访问层，所有sql查询都发生在其中。Adaper 代码是这样的：

from .models import *
class DBAdapter(object):
def __init__(self, session):
self._db_session = get_session() # get a sql alchemy session
def create_customer(**kwargs):
c = Customer(**kwargs)
self._db_session.add(c)
return c
.
.
.
def commit(self):
self.session.commit()
def close_session(self):
self.session.close()

该服务如下所示：

from ..adapter import DBAdapter
class Service1(object):
def __init__(self, a):
self.a = a
self._db_adapter = DBAdapter()
def do_something(self, x, y):
if x != 10:
self.create_something(x)
self._db_adapter.commit()
self._db_adapter.close_session()
return
self._db_adapter.create_customer(y)
self._db_adapter.create_something_else(x)
self._db_adapter.commit()
self._db_adapter.close_session()
return

现在的问题是如何关闭会话而不在每个 return 语句之前重复self._db_adapter.close_session()。它需要在两个地方关闭一次，然后返回内部，如果条件然后最后。

我是否应该以不同的方式创建数据访问层DBAdapter？

我想保持我的会话干净并仅在初始化服务时创建它，所以我不能像这样将其放在烧瓶中的 g 对象中：

烧瓶和炼金术：处理会话

Answer 1

让会话/事务充当上下文管理器，并从自定义/客户端程序中封装其行为：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from contextlib import contextmanager

# at the module level, the global sessionmaker,
# bound to a specific Engine
Session = sessionmaker(bind=engine)
class DBAdapter:
@contextmanager
def get_session(self):
session = Session()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()

<小时 />

from ..adapter import DBAdapter
from .models import *
class Service1:
def __init__(self, a):
self.a = a
self._db_adapter = DBAdapter()
def create_customer(session, **kwargs):
c = Customer(**kwargs)
session.add(c)
return c
def do_something(self, x, y):
if x != 10:
with self._db_adapter.get_session() as session:
self.create_something(session, x)
return None
with self._db_adapter.get_session() as session:
self.create_customer(session, y)
self.create_something_else(session, x)

Answer 2

Python 有类似内容的上下文管理器。考虑这样的事情：

import contextlib
@contextlib.contextmanager
def db_session():
db_adapter = None
try:
db_adapter = DBAdapter();
yield db_adapter
finally:
if db_adapter:
db_adapter.close_session()

现在，您可以自动关闭会话：

with db_session() as session:
customer = session.create_customer(...)
if something is wrong:
session.rollback()
return  # Exits the `with` scope, so closes session.
customer.set_name('Joe')
session.commit()
# Here session is closed.

或者，您可以考虑一种更方便的模式来为您提交/回滚。