SQLAlchemy：事务管理

Question

这是示例代码

engine = create_engine(url="postgresql+psycopg2://postgres:0000@localhost:5432/db")
SessionLocal = sessionmaker(autocommit = False, autoflush=False, bind=engine)

def get_database_session():
    try:
        session = SessionLocal()
        yield session
    finally:
        session.close()

app = FastAPI()

@app.post("/api/v1/users")
def add_user(data: AppUserCreateModel, session: Session = Depends(get_database_session)):
    with session.begin():
        #db call 1
        session.commit()

        #db call 2 results in an error
        session.commit()

我如何构建它，以便我可以在一个事务内进行多次提交，并且如果任何提交导致错误，它应该回滚所有以前的提交？

Answer 1

您将异常处理放在

get_database_session

函数中。另外，在 @python_user 注释的后面，一个“事务”通常应该保存所有 SQL 查询，并在最后进行一次提交。如果需要插入记录并获取主键值，可以使用

flush

方法。

def get_database_session():
    try:
        session = SessionLocal()
        yield session
    except Exception as err:
        #Would be better to limit this to specific SQLAlchemy errors
        session.rollback()
    finally:
        session.close()


@app.post("/api/v1/users")
def add_user(data: AppUserCreateModel, session: Session = Depends(get_database_session)):
    with session.begin() as db_session:
        # run you first set of queries
        db_session.flush()

        # run you second set of queries
        db_session.commit()