当前位置：首页 > news >正文

sqlalchemy 事务自动控制(类java aop)

news 来源：原创 2025/6/13 8:06:37

最近使用它交互数据库，想实现类似java aop那种自动事务控制，不用手动commit或者rollback。我是用的是flask+denpendency-injecter

这是我的db的配置类，里面会初始化一些session配置，里面比较重要的是把autocommit和autoflush关闭了，因为我们的代码会来处理这个，还有就是把expire_on_commit设置为flase，否则你commit之后，再取用某个entity就会报错了，例如你新建了一个entity，这个时候会更新他的id，返回给前端的时候就会报错了（Error Messages — SQLAlchemy 2.0 Documentation）。

"""Database module."""

from contextlib import contextmanager, AbstractContextManager
from typing import Callable

from sqlalchemy import create_engine, orm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
Base = declarative_base()

class DatabaseConfig:

    def __init__(self, db_url: str) -> None:
        self._engine = create_engine(db_url, echo=True)
        self._session_factory = orm.scoped_session(
            orm.sessionmaker(
                autocommit=False,
                autoflush=False,
                expire_on_commit=False,
                bind=self._engine,
            ),
        )

    def create_database(self) -> None:
        Base.metadata.create_all(self._engine)


    @contextmanager
    def session(self) -> Callable[..., AbstractContextManager[Session]]:
        session: Session = self._session_factory()
        try:
            yield session
        except Exception:
            session.rollback()
            raise
        else:
            if session._transaction.is_active:
                session.commit()
            session.close()

然后comtextmanger里面就是我们的处理代码了，我们主要依靠with代码块来控制，在yield之前的属于__init__，在yield之后属于__exit__，也就是当with代码块结束之前，如果发生任何报错，我们都会进行rollback操作，并且raise(这部分需要error handler来做了，这里就不赘述了)，然后如果什么错误都没有发生，就检测transaction是否还是active，如果是就commit，然后关闭session。

然后在Container中注入session contextmanager。

class Container(containers.DeclarativeContainer):
  
    wiring_config = containers.WiringConfiguration(packages=[
        "main"
    ])
    config = providers.Configuration(yaml_files=["config.yml"])
 
    db=providers.Singleton(DatabaseConfig,db_url=config.db.url)
    
    user_repository = providers.Factory(
        UserRepositoryImpl
    )
    
    user_service = providers.Factory(
        UserService,
        user_repository=user_repository,
        session_factory=db.provided.session
    )

然后再service层使用with代码块控制transation ，整个逻辑包含在同一个with中就行了。

class UserService:

    @inject
    def __init__(self, user_repository: UserRepository, session_factory: Callable[..., AbstractContextManager[Session]]) -> None:
        self._repository: UserRepository = user_repository
        self.session_factory=session_factory

    def create_user(self,user) -> User:
        with self.session_factory() as session:
            return self._repository.add(
                session=session,
                user=user
            )

然后在repo里面写具体代码就行了

class UserRepositoryImpl(UserRepository):

    def __init__(self) -> None:
        pass

    def add(self, user,session):
        session.add(user)
        return user

关于测试怎么写，可以看我篇blog，里面也会讲到测试类的事务控制：
Flask+ Dependency-injecter+pytest 写测试类-CSDN博客