29 使用数据库之基本CRUD操作

在上一篇中，我们探讨了如何使用 SQLAlchemy 连接数据库。在本节中，我们将继续扩展我们的 FastAPI 应用程序，实施基本的 CRUD（创建、读取、更新、删除）操作。这些操作可以帮助我们有效地管理数据库中的数据。在接下来的内容中，我们将通过案例演示如何实现这些操作。

准备工作

在开始之前，请确保你已经安装了所需的依赖包。你可以使用以下命令安装 FastAPI 和 SQLAlchemy ：

pip install fastapi[all] sqlalchemy databases

我们还需要配置数据库连接。以下示例将使用 SQLite 数据库。

# database.py
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

创建模型

接下来，我们需要定义用于 CRUD 操作的数据库模型。例如，假设我们正在管理一个 用户 数据库表。

# 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, index=True)
    email = Column(String, unique=True, index=True)

    def __repr__(self):
        return f"<User(name={self.name}, email={self.email})>"

现在我们已经定义了 User 模型，接下来将创建 CRUD 操作。

CRUD 操作实现

CRUD 操作通常会集中在一个服务模块中。我们可以创建一个名为 crud.py 的文件来实现这些功能。

# crud.py
from sqlalchemy.orm import Session
from models import User

# 创建用户
def create_user(db: Session, name: str, email: str):
    db_user = User(name=name, email=email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# 获取用户
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

# 获取所有用户
def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

# 更新用户
def update_user(db: Session, user_id: int, name: str, email: str):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        user.name = name
        user.email = email
        db.commit()
        db.refresh(user)
    return user

# 删除用户
def delete_user(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

主应用程序

现在，让我们将这些操作与 FastAPI 结合。在 main.py 文件中，我们将定义相应的 API 端点。

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine, Base
import crud
from models import User

# 创建所有表
Base.metadata.create_all(bind=engine)

app = FastAPI()

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=User)
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    return crud.create_user(db=db, name=name, email=email)

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

@app.get("/users/", response_model=list[User])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_users(db=db, skip=skip, limit=limit)

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, name: str, email: str, db: Session = Depends(get_db)):
    user = crud.update_user(db=db, user_id=user_id, name=name, email=email)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.delete("/users/{user_id}", response_model=User)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.delete_user(db=db, user_id=user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

总结

在本篇文章中，我们实现了基本的 CRUD 操作来管理用户数据。我们定义了数据库模型、CRUD 操作以及如何使用 FastAPI 在 API 端点中调用这些操作。这样，你就能够通过 HTTP 请求进行数据的创建、读取、更新和删除。

下一篇将讨论如何实现 异步数据库操作，以提高应用的性能。请继续关注。