Clean Architecture+FastAPI中SQLAlchemy多模型事务处理方案咨询

云上老周

2026-7-10

Clean Architecture（FastAPI）下SQLAlchemy多模型事务的实现方案

我在FastAPI环境下基于Clean Architecture开发应用，遇到了SQLAlchemy多模型事务处理的问题——之前看到的示例都是单模型场景，于是我写了一个包含Place和Coordinate关联模型的演示应用（代码如下）。目前实现了两个事务处理用例，都用了单个Session，我觉得CreatePlaceUseCase1更简洁，CreatePlaceUseCase2有点繁琐，想了解处理多模型数据创建的事务还有哪些可行方案？

from fastapi import FastAPI, Depends
from sqlalchemy import sessionmaker, DeclarativeBase, ForeignKey, mapped_column
from sqlalchemy.orm import Mapped, relationship, Session
from pydantic import BaseModel

app = FastAPI() 

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=...) 

class Base(DeclarativeBase) :
    pass

class Place(Base) :
    __tablename__ = "place"
    id: Mapped[int] = mapped_column(primary_key=True) 
    name: Mapped[str]
    # 补充反向关联（原代码遗漏）
    coordinate: Mapped["Coordinate"] = relationship(back_populates="place")

class Coordinate(Base) :
    __tablename__ = "coordinate"
    id: Mapped[int] = mapped_column(primary_key=True) 
    longitude: Mapped[float]
    latitude: Mapped[float]
    place_id: Mapped[int] = mapped_column(ForeignKey("place.id")) 
    place: Mapped["Place"] = relationship(back_populates="coordinate") 

def get_session() :
    session = SessionLocal() 
    yield session
    session.close() 

class CreatePlace(BaseModel):
    name: str
    coordinates: list[float, float] 

class Repository:
    model = None

    def __init__(self, session: Session):
        self.session = session
        if self.model is None:
            raise ValueError("model is not specified") 

    def create(self, create_data: dict, commit: bool = False) :
         instance = self.model(**create_data) 
         self.session.add(instance) 
         if commit:
             self.session.commit() 
         return instance

class PlaceRepository(Repository):
    model = Place

class CoordinateRepository(Repository):
    model = Coordinate

class CreatePlaceUseCase1:
    def __init__(self, session: Session = Depends(get_session)) :
        self.session = session
    
    def create_place(self, create_data: CreatePlace) :
        coordinate = Coordinate(longitude=create_data.coordinates[0], latitude=create_data.coordinates[1]) 
        place = Place(name=create_data.name, coordinate=coordinate) 
        self.session.add(place) 
        self.session.commit()

class CreatePlaceUseCase2:
    def __init__(self, session: Session = Depends(get_session)) :
        self.session = session
        self.place_repo = PlaceRepository(session)
        self.coordinate_repo = CoordinateRepository(session) 
    
    def create_place(self, create_data: CreatePlace):
        coordinate = self.coordinate_repo.create({"longitude": create_data.coordinates[0], "latitude": create_data.coordinates[1]}, commit=False) 
        self.place_repo.create({"name": create_data.name, "coordinate": coordinate}, commit=True)

@app.post("/create1")
def create_place_1(create_data: CreatePlace, use_case: CreatePlaceUseCase1 = Depends()):
    use_case.create_place(create_data) 
    return "Place created"

@app.post("/create2")
def create_place_2(create_data: CreatePlace, use_case: CreatePlaceUseCase2 = Depends()):
    use_case.create_place(create_data) 
    return "Place created"

可行方案汇总

1. 利用SQLAlchemy级联关联（UseCase1优化版）

既然Place和Coordinate已经通过ORM关系绑定，SQLAlchemy支持级联持久化——只要正确配置关系，添加父模型时会自动处理子模型的入库，无需单独操作子模型的Session。

优化要点：

给Place补充完整的反向关系，并设置级联规则（确保删除Place时自动删除关联的Coordinate）：

class Place(Base):
    __tablename__ = "place"
    id: Mapped[int] = mapped_column(primary_key=True) 
    name: Mapped[str]
    coordinate: Mapped["Coordinate"] = relationship(
        back_populates="place", 
        cascade="all, delete-orphan"
    )

用例层简化事务逻辑，添加异常回滚：

class CreatePlaceUseCase1:
    def __init__(self, session: Session = Depends(get_session)) :
        self.session = session
    
    def create_place(self, create_data: CreatePlace) :
        coordinate = Coordinate(longitude=create_data.coordinates[0], latitude=create_data.coordinates[1]) 
        place = Place(name=create_data.name, coordinate=coordinate) 
        self.session.add(place)
        try:
            self.session.commit()
            self.session.refresh(place)  # 可选，获取数据库生成的ID等字段
        except Exception:
            self.session.rollback()
            raise

这种方式最简洁，完全利用ORM特性，符合Clean Architecture中用例层只聚焦业务逻辑的原则。

2. 仓库层分离数据访问与事务（UseCase2优化版）

如果坚持使用仓库模式，核心原则是仓库只做数据CRUD，不处理事务边界——把commit/rollback的控制权交给用例层，避免单个仓库的create方法绑定事务逻辑。

优化后的仓库：

class Repository:
    model = None

    def __init__(self, session: Session):
        self.session = session
        if self.model is None:
            raise ValueError("model is not specified")

    def create(self, create_data: dict):
        instance = self.model(**create_data)
        self.session.add(instance)
        return instance  # 返回实例，方便用例层做关联

用例层统一控制事务：

class CreatePlaceUseCase2:
    def __init__(self, session: Session = Depends(get_session)) :
        self.session = session
        self.place_repo = PlaceRepository(session)
        self.coordinate_repo = CoordinateRepository(session) 
    
    def create_place(self, create_data: CreatePlace):
        coordinate = self.coordinate_repo.create({
            "longitude": create_data.coordinates[0], 
            "latitude": create_data.coordinates[1]
        })
        self.place_repo.create({
            "name": create_data.name, 
            "coordinate": coordinate
        })
        try:
            self.session.commit()
        except Exception:
            self.session.rollback()
            raise

这种方式让仓库职责更单一，用例层负责业务逻辑和事务边界，适合复杂业务场景的扩展。

3. 封装通用事务上下文管理器

如果多个用例都需要事务控制，可以封装一个工具类，避免重复编写try/except/rollback代码：

from contextlib import contextmanager

@contextmanager
def transaction(session: Session):
    try:
        yield
        session.commit()
    except Exception:
        session.rollback()
        raise

用例层使用：

class CreatePlaceUseCase3:
    def __init__(self, session: Session = Depends(get_session)) :
        self.session = session
        self.place_repo = PlaceRepository(session)
        self.coordinate_repo = CoordinateRepository(session) 
    
    def create_place(self, create_data: CreatePlace):
        with transaction(self.session):
            coordinate = self.coordinate_repo.create({
                "longitude": create_data.coordinates[0], 
                "latitude": create_data.coordinates[1]
            })
            self.place_repo.create({
                "name": create_data.name, 
                "coordinate": coordinate
            })

这种方式符合DRY原则，适合多场景复用事务逻辑。

4. 请求级全局事务（适合小型应用）

可以把事务控制放到Session依赖中，让每个请求自动包裹在事务里：

def get_session_with_transaction():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

之后用例或路由直接使用这个依赖，整个请求的所有数据库操作都会在同一个事务中完成，无需手动处理commit/rollback。但注意：这种方式适合请求级别的事务，如果一个请求包含多个独立业务操作，可能会导致事务范围过大。

方案选择建议

简单关联模型：优先选方案1，利用SQLAlchemy级联特性，代码最简洁。
复杂业务+仓库模式：选方案2或方案3，分离数据访问与事务控制，符合Clean Architecture分层原则。
小型应用/请求级事务：选方案4，减少重复代码。

内容的提问来源于stack exchange，提问作者Альберт Александров

Clean Architecture+FastAPI中SQLAlchemy多模型事务处理方案咨询

可行方案汇总 #

1. 利用SQLAlchemy级联关联（UseCase1优化版） #

2. 仓库层分离数据访问与事务（UseCase2优化版） #

3. 封装通用事务上下文管理器 #

4. 请求级全局事务（适合小型应用） #