python-3.x 在给定的项目结构中，管理 flask 应用中的sqlalchemy会话的最佳方式是什么？[duplicate]

osh3o9ms 于 2023-01-22 发布在 Python

关注(0)|答案(2)|浏览(87)

- 此问题在此处已有答案**：

Avoiding boilerplate session handling code in sqlalchemy functions（4个答案）
How to manage this commit/rollback pattern in sqlalchemy（2个答案）
三年前关闭了。
嗨，对不起，这是一个非常开放式的问题，更像是一个编码建议。寻找什么是使用sqlalchemy会话的 flask 项目的最佳实践。找不到一个博客，可以回答我的问题，因此在这里问它。
我有一个 flask 应用程序的项目结构类似如下：

---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

现在的问题是如何在每个return语句之前不重复self._db_adapter.close_session()的情况下关闭会话，需要在return之前在if条件中关闭两次，然后在结束时关闭。
我是否应该以不同的方式创建数据访问层DBAdapter？
我想保持会话干净，只在服务初始化时创建它，所以我不能像这样把它放在flask的g对象中：
Flask and sqlalchemy: handling sessions

python-3.x

来源：https://stackoverflow.com/questions/57060776/what-is-the-best-way-to-manage-sqlalchemy-sessions-in-a-flask-app-in-the-given-p

2条答案

按热度按时间

8qgya5xd1#

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.

或者您可以考虑使用even more convenient pattern，它可以为您执行提交/回滚。

赞(0）回复(0）举报 2023-01-22

qzwqbdag2#

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

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)

赞(0）回复(0）举报 2023-01-22

我来回答

python-3.x 在给定的项目结构中，管理 flask 应用中的sqlalchemy会话的最佳方式是什么？[duplicate]

2条答案

相关问题

热门标签

最新问答