FastAPI+SQLAlchemy基本操作

INFO

Learning Video: 知了传课【2026 版】零基础 FastAPI+Vue3+LangChain 实战

Python Version: 3.11 [miniconda]

OS: Ubuntu 24.04.3 LTS

IDE: Visual Studio Code

Shell: zsh 5.9

Install

bash
pip install "sqlalchemy[asyncio]" # ORM
pip install aiomysql  # 异步mysql支持
pip install cryptography  # 密码加密

创建连接

配置参数

python
DB_URL = "mysql+aiomysql://<username>:<password>@<host>:<port>/<dbname>?charset=utf8mb4"

创建 Engine 对象

创建异步 Engine 对象放置在/models/__init__.py中

python
from sqlalchemy.ext.asyncio import create_async_engine

DB_URL = "..."

engine = create_async_engine(
    DB_URL,
    echo=True,    # 开发阶段输出SQL执行日志
    pool_size=10,  # 连接池大小(d=5)
    max_overflow=20,  # 允许连接池最大的连接数(d=10)
    pool_timeout=10,  # 获取连接池连接的超时时间(d=30)
    pool_recycle=3600,  # 连接回收时间(d=-1，代表永不回收)
    pool_pre_ping=True  # 连接前是否预检查(d=False)
)

创建会话工厂

python
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession

AsyncSessionFactory = sessionmaker(
    bind=engine,  # 引擎或其子类对象
    class_=AsyncSession,  # Session类的代替
    autoflush=True,  # 是否在查找以前执行flush操作(d=True)
    expire_on_commit=False,  # 是否在提交commit操作后session就过期(d=True)
)

创建模型

定义 Base 类

Base类是所有ORM MODEL类的父类，一个ORM MODEL对应数据库中的一张表，ORM MODEL中的一个Column类属性对应数据库表中的一个字段

Base类的生成由函数 sqlalchemy.ext.declarative.declarative_base()实现，也可以继承sqlalchemy.MetaData类，实现紫的集子类以下为常规写法, 直接复制在init.py__中,models 中的其他类直接from . import BaseORM即可

python
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import MetaData

class BaseORM(DeclarativeBase):
    metadata = MetaData(
        naming_convention={
            # ix: index索引
            "ix": "ix_%(column_0_label)s",
            # uq: unique唯一索引
            "uq": "uq_%(table_name)s_%(column_0_name)s",
            # ck: Check检查约束
            "ck": "ck_%(table_name)s_%(constraint_name)s",
            # fk: ForeignKey外键索引
            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
            # pk: PrimaryKey主键约束
            "pk": "pk_%(table_name)s"
        }
    )

创建 ORM 模型

python
from typing import Optional
from sqlalchemy import Integer, String, select
from sqlalchemy.orm import Mapped, mapped_column


class User(BaseORM):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, auto_increment=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(100))
    password: Mapped[str] = mapped_column(String(200))

__tablename__: 表名
primary_key: 主键
unique: 唯一索引(内容不重复)
index: 索引

写法：<cname>: Mapped[<PyType>] = mapped_column(<SQLType>, <SQLOption>)

模型关系

python
from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey

外键：指向他表的 Column

python
user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"))

关系：用来跳转的属性

python
user: Mapped["User"] = relationship(back_populates="user_extention")

其中back_populates属性，表示反向引用的属性，即与另一个模型之间相互 relationship 的属性名

relationship中第一个参数为关联的模型，由于前面已经定义，可省略 user: Mapped["User"] = relationship(back_populates="user_extention")

和

user: Mapped["User"] = relationship("User", back_populates="user_extention") [recommend]

含义相同

一对一

Eg: 用户-扩展信息 User <-> UserExtention

python
class User(BaseORM):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(100))
    password: Mapped[str] = mapped_column(String(200))

    user_extention: Mapped["UserExtention"] = relationship(back_populates="user", uselist=False)

class UserExtention(BaseORM):
    __tablename__ = "user_extention"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    university: Mapped[str] = mapped_column(String(100))
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"), unique=True)
    user: Mapped["User"] = relationship(back_populates="user_extention")

声明back_populates的意义：
- user_extention.user 正向引用
- user.user_extention 反向引用
uselist=False(d=True): 正向引用保证一对一关系
unique=True: 对应到的 table 中保证一对一关系

一对多

Eg: 用户-文章 User <-> Article

python
class User(BaseORM):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(100))
    password: Mapped[str] = mapped_column(String(200))

    articles: Mapped[List["Article"]] = relationship(back_populates="author") # type: ignore

class Article(BaseORM):
    __tablename__ = "article"

    id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(100))
    content: Mapped[str] = mapped_column(String(1000))

    author_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"))
    author: Mapped["User"] = relationship(back_populates="articles") # type: ignore

多对多

Eg: 文章-标签 Article <-> Tag

通过中间表实现多对多关系

python
class Article(BaseORM):
    __tablename__ = "article"

    id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(100))
    content: Mapped[str] = mapped_column(String(1000))

    author_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"))
    author: Mapped["User"] = relationship(back_populates="articles") # type: ignore
    tags: Mapped[List["Tag"]] = relationship(
        "Tag",
        back_populates="articles",
        # 中间表
        secondary="article_tag"
    )


class Tag(BaseORM):
    __tablename__ = "tag"

    id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    articles: Mapped[List["Article"]] = relationship(
        "Article",
        back_populates="tags",
        # 中间表
        secondary="article_tag"
    )

class ArticleTag(BaseORM):
    __tablename__ = "article_tag"

    # 两个主键
    article_id: Mapped[int] = mapped_column(Integer, ForeignKey("article.id"), primary_key=True)
    tag_id: Mapped[int] = mapped_column(Integer, ForeignKey("tag.id"), primary_key=True)

模型迁移

安装 alembic

bash
pip install alembic

流程

创建迁移仓库
修改 alembic.ini
修改 env.py
生成迁移脚本
执行迁移脚本

创建迁移仓库

只需要进行一次

bash
alembic init alembic --template async

修改 alembic.ini

只需要进行一次

注释掉alembic.ini中的sqlalchemy.url

修改 env.py

只需要进行一次

alembic/env.py中添加内容

python
config = context.config
# 下方开始添加内容

# ============== USER DATABASE ================

database_url = <YOUR_DB_URL>
if database_url is None:
    raise ValueError("Database URL is not set")
config.set_main_option("sqlalchemy.url", database_url)

# =============================================

<YOUR_DB_URL>是数据库连接地址，通过你的位置 import 导入

修改alembic/env.py中的target_meatadata

target_metadata = None修改为

python
target_metadata = BaseORM.metadata

BaseORM是自定义的数据库模型基类

生成迁移脚本

DB、Table 变化后需要执行

执行前请确保models/__init_.py中 import 了所有的 py 模块

写在最后：

python
# 导入ORM
from . import user, article

随后执行

bash
alembic revision --autogenerate -m "<message>"

执行迁移脚本

DB、Table 变化后需要执行

bash
alembic upgrade head

CRUD 操作

增删改查操作

创建 Session 对象

使用异步上下文管理器
使用依赖注入 [Recommend]
使用中间件

使用依赖注入

python
async def get_session() -> AsyncSession:
    session = AsyncSessionFactory()  # 这里使用的是工厂
    try:
        yield session
    finally:
        await session.close()

@app.get("/article/add")
async def add_article(
    req: UserCreateReq,
    # 这里使用的是Session类
    session: AsyncSession = Depends(get_session)
):
    async with session.begin():
        user = User(
            username=req.username,
            email=req.email,
            password=req.password
        )
        session.add(user)
    return user

这里可以创建一个 schemas 模块，用于定义请求参数和响应参数可以指定 POST 回给前端的数据类

python
class UserSchemaOut(BaseModel):
    # 返回给前端创建成功的信息
    id: int
    username: str
    email: str

@app.post("/user/add", response_model=UserSchemaOut)
async def add_user(
    user: UserCreateReq,
    session: AsyncSession = Depends(get_session)
):
    async with session.begin():
        user = User(
            username=req.username,
            email=req.email,
            password=req.password
        )
        session.add(user)
    return user

增删改查

新增数据

python
@app.post(...)
async def add_data(...):
    async with session.begin():
        data = ...
        session.add(data)
    return data

删除数据

python
from sqlalchemy import delete as DELETE

python
@app.delete("/XXX/delete/{XXX_id}")
async def add_data():
    async with session.begin():
        await session.execute(
            DELETE(<TableORM>).where(<Condition>)
        )
    return {"message": "Delete Success"}

查询数据

查找一条数据

python
from sqlalchemy import select as SELECT

python
@app.get("/user/delete/{user_id}", response_model=UserSchemaOut)
async def add_data(
    user_id: int,
    session: As = Depends(get_session)
):
    async with session.begin():
        query = await session.execute(
            SELECT(
                User.id, User.email, User.username
            ).where(
                User.id == user_id
            )
        )
        result = query.scalar()  # 返回一条数据
    # 可选：无数据404
    if result is None:
        raise HTTPException(status_code=404, detail="Data Not Found")
    return result

这里的 query 可以拆分，execute 也可以直接用后面的 scalar

python
    async with session.begin():
        stmt = SELECT(
            User.id, User.email, User.username
        ).where(
            User.id == user_id
        )
        result = await session.scalar(stmt)
    return result

查找多条数据

python
@app.get("/user/select/qq", response_model=UserListSchemaOut)
async def get_user(
    session: AsyncSession = Depends(get_session)
):
    async with session.begin():
        stmt = SELECT(User).where(
            User.email.contains("qq.com")
        )
        users = await session.scalars(stmt)
        return users

修改数据

python
from sqlalchemy import update as UPDATE

两步修改

适合于需要查找信息的场景

python
@app.put("/user/update/{user_id}", response_model=UserSchemaOut)
async def update_user(
    user_id: int,
    session: AsyncSession = Depends(get_session)
):
    # 1.先查再改
    async with session.begin():
        user = await session.scalar(
            SELECT(User).where(User.id==user_id)
        )
        if user is None:
            raise HTTPException(status_code=404, detail=f"User_{user_id} not found")
        user.username = user.username+"-[BAN]"
    return user

直接修改

适合于不需要查找信息且没有返回信息的场景

python
@app.put("/user/update/{user_id}")
async def update_user(
    user_id: int,
    session: AsyncSession = Depends(get_session)
):
    async with session.begin():
        await session.execute(
            UPDATE(User).where(User.id==user_id).values(
                username="[BAN]"
            )
        )
        return {"message": "Update Success"}

目录

Install

创建连接

配置参数

创建 Engine 对象

创建会话工厂

创建模型

定义 Base 类

创建 ORM 模型

模型关系

一对一

一对多

多对多

模型迁移

安装 alembic

流程

创建迁移仓库

修改 alembic.ini

修改 env.py

生成迁移脚本

执行迁移脚本

CRUD 操作

创建 Session 对象

使用依赖注入

增删改查

新增数据

删除数据

查询数据

查找一条数据

查找多条数据

修改数据

两步修改

直接修改