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

Question

您好，抱歉，这是一个非常开放式的问题，更像是编码建议。寻找在 Flask 项目中使用 sqlalchemy 会话的最佳实践。找不到可以回答我的问题的博客，因此在这里提问。

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

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

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

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

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

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

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

我想保持会话干净并仅在初始化服务时创建它，因此我不能将其放入 Flask 中的 g 对象中，如下所示：

Flask 和 sqlalchemy：处理会话

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)