FastAPI 与数据库交互教程

FastAPI 是一个快速（高性能）的 Web 框架，它能够帮助我们快速搭建 Web 应用，而数据库则是数据持久化存储的重要工具。今天，我们就来学习如何使用 FastAPI 与数据库进行交互，这里我们选择使用 SQLite 数据库，它是一种轻量级的数据库，非常适合初学者和小型项目。

环境准备

在开始之前，我们需要确保已经安装了 FastAPI 和 SQLAlchemy（一个用于 Python 的强大的数据库抽象层库）。可以使用以下命令进行安装：

pip install fastapi uvicorn sqlalchemy

这里 uvicorn 是一个 ASGI 服务器，用于运行 FastAPI 应用。

项目结构

我们创建一个简单的项目结构，在项目根目录下创建以下文件：

/
├── main.py
├── models.py
├── schemas.py

schemas.py 文件
这个文件主要用于定义数据的结构和验证规则，使用 Pydantic 库。Pydantic 可以帮助我们对数据进行验证和序列化。

# schemas.py
from pydantic import BaseModel

# 定义一个用户信息的模型，用于接收和验证客户端传来的数据
class UserInfo(BaseModel):
    name: str  # 用户姓名
    role: str  # 用户角色，例如“山海摸鱼人”等

    class Config:
        orm_mode = True  # 支持从 ORM 对象中读取数据

models.py 文件
这个文件用于定义数据库表的结构，使用 SQLAlchemy 库。

# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# 创建一个基类，所有的数据库模型类都将继承自这个基类
Base = declarative_base()

# 定义一个用户表的模型
class User(Base):
    __tablename__ = "users"  # 表名

    id = Column(Integer, primary_key=True, index=True)  # 用户 ID，主键，用于唯一标识用户
    name = Column(String)  # 用户姓名
    role = Column(String)  # 用户角色，例如“山海摸鱼人”等

main.py 文件
这个文件是我们的主应用文件，用于创建 FastAPI 应用、配置数据库连接和定义路由。

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base, User
from schemas import UserInfo

# 创建 FastAPI 应用实例
app = FastAPI()

# 数据库连接字符串，使用 SQLite 数据库，数据库文件名为 test.db
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.metadata.create_all(bind=engine)

# 定义一个依赖函数，用于获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 定义一个创建用户的路由
@app.post("/users/")
def create_user(user: UserInfo, db: sessionmaker = Depends(get_db)):
    # 创建一个新的用户对象
    db_user = User(name=user.name, role=user.role)
    # 将用户对象添加到数据库会话中
    db.add(db_user)
    # 提交数据库会话，将数据保存到数据库中
    db.commit()
    # 刷新数据库会话，获取数据库中最新的用户信息
    db.refresh(db_user)
    return db_user

# 定义一个获取所有用户的路由
@app.get("/users/")
def read_users(skip: int = 0, limit: int = 100, db: sessionmaker = Depends(get_db)):
    # 从数据库中查询用户信息
    users = db.query(User).offset(skip).limit(limit).all()
    return users

代码解释

schemas.py 文件
UserInfo 类继承自 BaseModel，它定义了用户信息的结构，包含 name 和 role 两个字段。orm_mode = True 表示这个模型可以和 ORM 对象进行交互。

models.py 文件

• Base = declarative_base() 创建了一个基类，所有的数据库模型类都将继承自这个基类。
• User 类继承自 Base，定义了一个名为 users 的数据库表，包含 id、name 和 role 三个字段。

main.py 文件

• app = FastAPI() 创建了一个 FastAPI 应用实例。
• SQLALCHEMY_DATABASE_URL 定义了数据库连接字符串，这里使用 SQLite 数据库。
• engine = create_engine(...) 创建了一个数据库引擎，用于连接数据库。
• SessionLocal = sessionmaker(...) 创建了一个会话工厂，用于创建数据库会话。
• Base.metadata.create_all(bind=engine) 创建了数据库表。
• get_db 函数是一个依赖函数，用于获取数据库会话，并在使用完后关闭会话。
• create_user 路由用于创建新用户，接收一个 UserInfo 对象和一个数据库会话，将用户信息保存到数据库中，并返回保存后的用户信息。
• read_users 路由用于获取所有用户信息，接收 skip 和 limit 两个参数，用于分页查询，返回查询到的用户信息。

运行与测试

在终端中运行以下命令启动 FastAPI 应用：

uvicorn main:app --reload

这里 main 是主应用文件的文件名，app 是 FastAPI 应用实例的名称，--reload 表示开启热重载，当代码发生变化时，应用会自动重启。

可以使用工具如 Postman 或浏览器来测试 API。

创建用户
使用 Postman 发送一个 POST 请求到 http://127.0.0.1:8000/users/，请求体如下：

{
    "name": "张三",
    "role": "摸鱼人"
}

响应结果会返回保存后的用户信息，包含用户的 id、name 和 role。

获取所有用户
在浏览器中访问 http://127.0.0.1:8000/users/，会返回数据库中所有用户的信息。使用 MySQL 数据库

如果要将数据库从 SQLite 更换为 MySQL 时，需要对代码中的数据库连接部分进行修改。 安装 pymysql 作为 MySQL 的驱动程序：

pip install pymysql

在 main.py 文件开头添加对 pymysql 的导入：

import pymysql

将 SQLite 的数据库连接字符串替换为 MySQL 的连接字符串。MySQL 的连接字符串格式通常为 mysql+pymysql://username:password@host:port/database_name。假设你的 MySQL 用户名是 root，密码是 password，主机是 localhost，端口是 3306，数据库名是 testdb，则连接字符串如下：

# 修改数据库连接字符串为 MySQL 的连接字符串
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:password@localhost:3306/testdb"

由于使用的数据库不同，创建数据库引擎时不需要 connect_args={"check_same_thread": False} 这个参数了，因为 MySQL 没有这个线程检查的限制。修改后的代码如下：

# 创建数据库引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL)

注意事项

数据库创建：在运行代码之前，你需要确保 MySQL 中已经创建了相应的数据库。可以使用 MySQL 的命令行工具或者可视化工具（如 Navicat）来创建数据库。例如，在 MySQL 命令行中执行 CREATE DATABASE testdb; 来创建名为 testdb 的数据库。

字符编码：如果涉及到中文等非 ASCII 字符，建议在创建数据库时指定字符编码为 utf8mb4，以避免字符编码问题。例如，在创建数据库时可以使用 CREATE DATABASE testdb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;。

通过以上步骤，你就可以将 FastAPI 应用从使用 SQLite 数据库切换到使用 MySQL 数据库了。

来源：山海摸鱼人