FastAPI+SQLite：快速搭建轻量级数据库API服务

在现代应用开发中，快速构建数据库驱动的API服务是常见需求。FastAPI作为高性能Python框架，结合SQLite这一轻量级嵌入式数据库，能帮助我们快速实现简单但实用的后端服务。本文将通过一个完整的“学生信息管理”示例，手把手带你从0到1搭建一个数据库API服务。

一、环境准备：安装必要工具¶

首先，我们需要安装FastAPI、SQLite的Python接口（SQLAlchemy，用于ORM操作）和服务器（Uvicorn）。打开终端，执行以下命令：

pip install fastapi uvicorn sqlalchemy

FastAPI：Python高性能Web框架，自动生成API文档，支持异步操作。
Uvicorn：ASGI服务器，用于运行FastAPI应用。
SQLAlchemy：ORM（对象关系映射）库，让Python代码可直接操作数据库，无需写原生SQL。
SQLite：无需独立服务器的嵌入式数据库，适合开发和小型项目。

二、核心概念：数据模型与数据库连接¶

我们以“学生信息”为例，先定义数据模型和数据库连接。创建一个main.py文件，按以下步骤编写代码：

1. 导入依赖¶

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

2. 定义数据模型¶

使用SQLAlchemy定义数据库表结构（ORM模型），这里我们创建一个Student模型，对应数据库中的students表：

# 1. 连接SQLite数据库（文件存储，无需额外服务器）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 数据库文件路径为当前目录下的test.db

# 2. 创建SQLAlchemy引擎（核心连接对象）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite专用参数
)

# 3. 创建会话本地类（用于操作数据库）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 4. 定义基类（所有ORM模型继承自它）
Base = declarative_base()

# 5. 定义Student模型（对应数据库表）
class Student(Base):
    __tablename__ = "students"  # 表名

    id = Column(Integer, primary_key=True, index=True)  # 主键，自增
    name = Column(String, index=True, nullable=False)   # 学生姓名（非空）
    age = Column(Integer, nullable=True)               # 学生年龄（可选）

3. 创建数据库表¶

在第一次运行时，需要创建表结构。可以通过SQLAlchemy的create_all()方法自动生成表：

# 创建所有表（需在定义完模型后执行）
Base.metadata.create_all(bind=engine)

4. 定义Pydantic模型（数据验证与序列化）¶

FastAPI需要明确请求/响应的数据格式，我们用Pydantic定义数据模型：

# 用于请求数据的模型（无需id，由数据库自动生成）
class StudentCreate(BaseModel):
    name: str
    age: int | None = None  # 年龄可选

# 用于响应数据的模型（包含id）
class StudentResponse(BaseModel):
    id: int
    name: str
    age: int | None = None

    class Config:
        orm_mode = True  # 允许从ORM对象（SQLAlchemy模型）中读取数据

三、数据库操作函数（CRUD）¶

接下来，编写操作数据库的函数（CRUD：创建、读取、更新、删除）。我们通过SQLAlchemy的会话（Session）执行数据库操作：

# 依赖项：获取数据库会话（每次请求生成新会话，用完关闭）
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 1. 创建学生（POST /students）
def create_student(db: Session, student: StudentCreate):
    db_student = Student(name=student.name, age=student.age)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)  # 刷新数据，获取数据库生成的id
    return db_student

# 2. 获取单个学生（GET /students/{student_id}）
def get_student(db: Session, student_id: int):
    return db.query(Student).filter(Student.id == student_id).first()

# 3. 获取所有学生（GET /students）
def get_all_students(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Student).offset(skip).limit(limit).all()

# 4. 更新学生（PUT /students/{student_id}）
def update_student(db: Session, student_id: int, student_update: StudentCreate):
    db_student = get_student(db, student_id)
    if not db_student:
        return None
    # 更新字段（仅name和age，id不可修改）
    db_student.name = student_update.name
    db_student.age = student_update.age
    db.commit()
    db.refresh(db_student)
    return db_student

# 5. 删除学生（DELETE /students/{student_id}）
def delete_student(db: Session, student_id: int):
    db_student = get_student(db, student_id)
    if not db_student:
        return False
    db.delete(db_student)
    db.commit()
    return True

四、搭建FastAPI路由（API端点）¶

现在，将CRUD函数与FastAPI路由绑定，实现具体的API接口：

# 创建FastAPI实例
app = FastAPI(title="学生信息管理API")

# 1. 创建学生（POST请求）
@app.post("/students", response_model=StudentResponse, status_code=201)
def create_student_endpoint(student: StudentCreate, db: Session = Depends(get_db)):
    db_student = create_student(db=db, student=student)
    return db_student

# 2. 获取单个学生（GET请求）
@app.get("/students/{student_id}", response_model=StudentResponse)
def get_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    db_student = get_student(db, student_id)
    if not db_student:
        raise HTTPException(status_code=404, detail="学生不存在")
    return db_student

# 3. 获取所有学生（GET请求）
@app.get("/students", response_model=list[StudentResponse])
def get_all_students_endpoint(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    students = get_all_students(db, skip=skip, limit=limit)
    return students

# 4. 更新学生（PUT请求）
@app.put("/students/{student_id}", response_model=StudentResponse)
def update_student_endpoint(student_id: int, student: StudentCreate, db: Session = Depends(get_db)):
    db_student = update_student(db, student_id, student)
    if not db_student:
        raise HTTPException(status_code=404, detail="学生不存在")
    return db_student

# 5. 删除学生（DELETE请求）
@app.delete("/students/{student_id}", status_code=204)
def delete_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    success = delete_student(db, student_id)
    if not success:
        raise HTTPException(status_code=404, detail="学生不存在")
    return None

五、运行与测试API服务¶

1. 启动应用¶

在终端执行以下命令启动服务（--reload开启热重载，修改代码后自动重启）：

uvicorn main:app --reload

main：文件名（main.py）
app：FastAPI实例名
启动后，访问 http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc 可查看交互式API文档。

2. 测试API¶

使用FastAPI的交互式文档（Swagger UI）或命令行工具测试：

创建学生：
点击 /docs 中的 POST /students → 点击“Try it out” → 输入JSON数据（如 {"name": "Alice", "age": 20}）→ 点击“Execute”。
获取单个学生：
点击 GET /students/{student_id} → 输入刚创建的学生ID（如 1）→ 点击“Execute”。
其他操作：
同理测试 GET /students（获取所有）、PUT /students/{id}（更新）、DELETE /students/{id}（删除）。

六、总结¶

通过本文，你已完成一个基于FastAPI+SQLite的轻量级数据库API服务。核心优势包括：

轻量易用：SQLite无需额外服务器，FastAPI自动生成API文档，降低开发门槛。
高性能：FastAPI支持异步操作，适合高并发场景（但SQLite本身是单线程，适合中小规模）。
扩展性强：可通过修改SQLAlchemy模型和路由，快速支持复杂业务（如多表关联、权限控制）。

后续可尝试扩展：添加分页、过滤、排序功能，或迁移到PostgreSQL/MySQL等数据库。

完整代码示例（可直接复制到main.py运行）：

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# 数据库配置
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()

# 学生模型（数据库表）
class Student(Base):
    __tablename__ = "students"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    age = Column(Integer, nullable=True)

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

# Pydantic模型（数据验证）
class StudentCreate(BaseModel):
    name: str
    age: int | None = None

class StudentResponse(BaseModel):
    id: int
    name: str
    age: int | None = None
    class Config:
        orm_mode = True

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

# 数据库操作函数
def create_student(db: Session, student: StudentCreate):
    db_student = Student(name=student.name, age=student.age)
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student

def get_student(db: Session, student_id: int):
    return db.query(Student).filter(Student.id == student_id).first()

def get_all_students(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Student).offset(skip).limit(limit).all()

def update_student(db: Session, student_id: int, student: StudentCreate):
    db_student = get_student(db, student_id)
    if not db_student:
        return None
    db_student.name = student.name
    db_student.age = student.age
    db.commit()
    db.refresh(db_student)
    return db_student

def delete_student(db: Session, student_id: int):
    db_student = get_student(db, student_id)
    if not db_student:
        return False
    db.delete(db_student)
    db.commit()
    return True

# FastAPI应用
app = FastAPI(title="学生信息API")

# API路由
@app.post("/students", response_model=StudentResponse, status_code=201)
def create_student_endpoint(student: StudentCreate, db: Session = Depends(get_db)):
    return create_student(db, student)

@app.get("/students/{student_id}", response_model=StudentResponse)
def get_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    student = get_student(db, student_id)
    if not student:
        raise HTTPException(status_code=404, detail="学生不存在")
    return student

@app.get("/students", response_model=list[StudentResponse])
def get_all_students_endpoint(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return get_all_students(db, skip, limit)

@app.put("/students/{student_id}", response_model=StudentResponse)
def update_student_endpoint(student_id: int, student: StudentCreate, db: Session = Depends(get_db)):
    updated = update_student(db, student_id, student)
    if not updated:
        raise HTTPException(status_code=404, detail="学生不存在")
    return updated

@app.delete("/students/{student_id}", status_code=204)
def delete_student_endpoint(student_id: int, db: Session = Depends(get_db)):
    success = delete_student(db, student_id)
    if not success:
        raise HTTPException(status_code=404, detail="学生不存在")
    return None