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

SQLModel与FastAPI结合：构建用户增删改查接口

news 2025/7/15 20:35:37

SQLModel简介

SQLModel是一个现代化的Python库，旨在简化与数据库的交互。它结合了Pydantic和SQLAlchemy的优势，使得定义数据模型、进行数据验证和与数据库交互变得更加直观和高效。SQLModel由FastAPI的创始人Sebastián Ramírez开发，专为与FastAPI框架无缝集成而设计。

SQLModel的优点

简洁性：通过结合Pydantic的数据验证和SQLAlchemy的ORM功能，SQLModel使模型定义和数据库操作更加简洁。
类型安全：充分利用Python的类型提示，增强代码的可读性和可靠性。
与FastAPI无缝集成：优化了与FastAPI的集成，支持自动文档生成和依赖注入。
灵活性：支持同步和异步操作，适应不同的性能需求。
现代化设计：采用现代化的Python编码风格和最佳实践，提升开发体验。

如何与FastAPI结合使用

FastAPI和SQLModel的结合为现代Web应用开发带来了一系列显著的优势，特别适合需要与SQL数据库交互的场景。以下是它们结合的一些主要优点：

简短的代码：SQLModel通过使用Python类型注解来定义数据模型，最小化代码重复，无需在SQLAlchemy和Pydantic之间复制模型。
简单易用：API设计简单，强大的编辑器支持，学习曲线低，可以快速上手。
可扩展性：拥有SQLAlchemy和Pydantic的所有功能，同时保持了代码的简洁性。
高性能：SQLModel采用了性能优化策略，如预编译SQL语句、减少数据库连接次数等，提高数据库操作性能。
支持异步操作：与asyncio库一起使用，提高高并发场景下的程序性能。
支持原生SQL：可以使用原生SQL语句进行数据库操作，同时支持参数绑定和SQL注入防护。

用户增删改查接口实例

以下是一个使用FastAPI和SQLModel实现用户增删改查（CRUD）操作的简单案例：

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import SQLModel, Field, create_engine, Session
from typing import Optional

# 定义数据库URL和创建数据库引擎
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)

# 定义用户模型
class UserBase(SQLModel):
    name: Optional[str] = None
    age: Optional[int] = None

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

# 创建数据库和表
SQLModel.metadata.create_all(engine)

# 依赖项，用于获取数据库会话
def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()

# 创建用户
@app.post("/users/")
def create_user(user: UserBase, db: Session = Depends(get_db)):
    db_user = User.from_orm(user)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 读取用户
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.get(User, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# 更新用户
@app.put("/users/{user_id}")
def update_user(user_id: int, user: UserBase, db: Session = Depends(get_db)):
    db_user = db.get(User, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db_user.name = user.name
    db_user.age = user.age
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 删除用户
@app.delete("/users/{user_id}", response_class=HTTPException)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.get(User, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}