FastAPI数据库集成与事务管理

灌篮少年 · 2025-4-18 00:27:22

title: FastAPI数据库集成与事务管理
date: 2025/04/18 00:15:34
updated: 2025/04/18 00:15:34
author: cmdragon
excerpt:
FastAPI与SQLAlchemy集成指南：首先配置SQLite数据库毗连，创建会话工厂和声明性基类。界说用户模子并映射表布局，使用Pydantic进行数据验证。通过依靠项获取数据库会话，实现用户创建和转账功能。事务控制通过显式开始事务和错误处理确保数据一致性。常见标题包括422验证错误、500服务器错误和完整性错误，需检查哀求体、数据库毗连和束缚。课后Quiz解答事务回滚、SQL注入防护和并发写操作处理。
categories:

后端开发
FastAPI

tags:

FastAPI
SQLAlchemy
数据库集成
事务管理
Pydantic
错误处理
并发控制

扫描二维码关注大概微信搜一搜：编程智域前端至全栈交换与发展
探索数千个预构建的 AI 应用，开启你的下一个巨大创意
FastAPI数据库集成与事务管理完全指南

1. 环境预备与底子配置

在项目根目录创建database.py文件：

# 安装依赖：pip install fastapi uvicorn sqlalchemy pydantic
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 配置SQLite数据库连接（生产环境建议使用PostgreSQL）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
echo=True # 显示生成的SQL语句
)
# 创建数据库会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 声明性基类
Base = declarative_base()

复制代码

2. 模子界说与表布局映射

from sqlalchemy import Column, Integer, String
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True)
balance = Column(Integer, default=0)
# 创建数据库表（生产环境建议使用迁移工具）
Base.metadata.create_all(bind=engine)

复制代码

3. Pydantic数据模子

from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
balance: int = 0
class UserResponse(BaseModel):
id: int
name: str
email: str
balance: int
class Config:
orm_mode = True # 启用ORM模式转换

复制代码

4. 路由与数据库操作

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
router = APIRouter()
# 依赖项获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
# 显式开始事务
transaction = db.begin()
try:
db_user = User(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
except Exception as e:
transaction.rollback()
raise HTTPException(
status_code=400,
detail=f"创建用户失败: {str(e)}"
)

复制代码

5. 事务控制与错误处理

多操作事务示例：

def transfer_funds(sender_id: int, receiver_id: int, amount: int, db: Session):
transaction = db.begin()
try:
# 获取发送方账户
sender = db.query(User).filter(User.id == sender_id).with_for_update().first()
if not sender or sender.balance < amount:
raise ValueError("余额不足或账户不存在")
# 获取接收方账户
receiver = db.query(User).filter(User.id == receiver_id).with_for_update().first()
if not receiver:
raise ValueError("接收方账户不存在")
# 执行转账
sender.balance -= amount
receiver.balance += amount
db.commit()
return {"message": "转账成功"}
except Exception as e:
transaction.rollback()
raise HTTPException(status_code=400, detail=str(e))

复制代码

6. 常见报错解决方案

标题1：422 Validation Error

现象：哀求参数验证失败
解决方法：
- 检查哀求体是否符合Pydantic模子界说
- 使用Swagger UI测试接口
- 查看返回的detail字段中的具体错误信息

标题2：500 Internal Server Error

现象：数据库毗连失败
解决方法：
- 检查数据库URL格式是否正确
- 验证数据库服务是否正常运行
- 检查数据库用户权限设置

标题3：IntegrityError (sqlalchemy.exc.IntegrityError)

现象：违反数据库束缚
解决方法：
- 检查唯一性束缚字段（如email）
- 验证外键关联是否存在
- 确保NOT NULL字段都有值

		自动登录	找回密码
密码			立即注册

FastAPI数据库集成与事务管理

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块