FastAPI与SQLAlchemy数据库集成与CRUD操纵

渣渣兔 · 2025-4-16 10:57:31

title: FastAPI与SQLAlchemy数据库集成与CRUD操纵
date: 2025/04/16 09:50:57
updated: 2025/04/16 09:50:57
author: cmdragon
excerpt:
FastAPI与SQLAlchemy集成基础包括环境准备、数据库毗连设置和模型定义。CRUD操纵通过数据访问层封装和路由层实现，确保线程安全和事务管理。常见错误如422哀求验证错误通过Pydantic模型和中间件处理。Session生命周期管理依赖注入系统保证每个哀求独立会话。常见报错如数据库毗连失败和事务回滚通过查抄服务状态、验证毗连参数和非常处理解决。
categories:

后端开辟
FastAPI

tags:

FastAPI
SQLAlchemy
数据库集成
CRUD操纵
Session管理
错误处理
MySQL

扫描二维码关注或者微信搜一搜：编程智域前端至全栈互换与成长
探索数千个预构建的 AI 应用，开启你的下一个伟大创意
1. FastAPI 与 SQLAlchemy 同步数据库集成基础

1.1 环境准备与安装

首先创建虚拟环境并安装必要依赖：

python -m venv venv
source venv/bin/activate # Linux/Mac
venv\Scripts\activate.bat # Windows
pip install fastapi uvicorn sqlalchemy pymysql

复制代码

1.2 数据库毗连设置

在database.py中设置焦点数据库毗连：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://user:password@localhost/mydatabase"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=20,
max_overflow=0,
pool_pre_ping=True
)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine,
expire_on_commit=False
)

复制代码

1.3 模型定义与关系映射

在models.py中定义数据模型：

from sqlalchemy import Column, Integer, String
from database import Base
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)
age = Column(Integer, default=18)
def __repr__(self):
return f"<User(name='{self.name}', email='{self.email}')>"

复制代码

2. CRUD 操纵标准实现模式

2.1 数据访问层封装

创建repository.py实现CRUD操纵：

from sqlalchemy.orm import Session
from models import User
class UserRepository:
@staticmethod
def create_user(db: Session, user_data: dict):
""" 创建用户 """
db_user = User(**user_data)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@staticmethod
def get_user(db: Session, user_id: int):
""" 根据ID获取用户 """
return db.query(User).filter(User.id == user_id).first()
@staticmethod
def update_user(db: Session, user_id: int, update_data: dict):
""" 更新用户信息 """
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
for key, value in update_data.items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
@staticmethod
def delete_user(db: Session, user_id: int):
""" 删除用户 """
db_user = db.query(User).filter(User.id == user_id).first()
if db_user:
db.delete(db_user)
db.commit()
return True
return False

复制代码

2.2 路由层实现

在main.py中定义API端点：

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from models import Base
from database import engine, SessionLocal
from repository import UserRepository
from pydantic import BaseModel
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 依赖注入获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
class UserCreate(BaseModel):
name: str
email: str
age: int = 18
@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
return UserRepository.create_user(db, user.dict())
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = UserRepository.get_user(db, user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user

复制代码

3. Session 生命周期管理

3.1 Session 线程安全策略

通过依赖注入系统保证每个哀求独立会话：

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

复制代码

3.2 事务管理最佳实践

def transfer_funds(db: Session, from_id: int, to_id: int, amount: float):
try:
from_user = UserRepository.get_user(db, from_id)
to_user = UserRepository.get_user(db, to_id)
if from_user.balance < amount:
raise ValueError("Insufficient funds")
from_user.balance -= amount
to_user.balance += amount
db.commit()
except Exception as e:
db.rollback()
raise e

复制代码

4. 常见错误处理

4.1 422 哀求验证错误

示例场景：

{
"detail": [
{
"loc": [
"body",
"age"
],
"msg": "value is not a valid integer",
"type": "type_error.integer"
}
]
}

复制代码

解决方案：

查抄哀求体是否匹配Pydantic模型定义
使用OpenAPI文档进行测试
添加中间件捕获验证错误：

from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={"detail": exc.errors(), "body": exc.body},
)

复制代码

课后Quiz

问题1：以下哪种方式可以有效防止SQL注入攻击？
A) 使用字符串拼接SQL语句
B) 使用ORM的查询参数化功能
C) 在数据库毗连字符串添加特别参数
D) 禁用所有输入验证
答案：B) 正确。SQLAlchemy等ORM框架会自动进行参数化查询，将用户输入作为参数传递而不是直接拼接到SQL语句中。
问题2：为什么需要在finally块中关闭数据库会话？
A) 为了提升查询性能
B) 确保会话在任何情况下都会被正确关闭
C) 防止其他哀求使用该会话
D) 满意数据库毗连池的要求
答案：B) 正确。无论是否发生非常，finally块中的代码都会执行，保证会话资源的正确开释。
常见报错解决方案

**报错1
**：sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2003, "Can't connect to MySQL server on 'localhost'")
缘故原由分析：

数据库服务未启动
毗连参数（用户名/密码/端口）错误
网络防火墙阻止毗连

解决方案：

查抄MySQL服务状态
验证毗连字符串参数
使用telnet测试端口连通性
添加毗连超时参数：

create_engine(connect_args={"connect_timeout": 10})

复制代码

**报错2
**：sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back due to a previous exception during flush.
缘故原由分析：

数据库操纵违背约束（如唯一性约束）
事务未正确处理非常

解决方案：

查抄数据完整性约束
在事务代码块中添加try/except
执行显式回滚操纵
使用session.expire_all()重置会话状态

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

FastAPI与SQLAlchemy数据库集成与CRUD操纵

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块