FastAPI连接MySQL实现自动建表的示例代码_Mysql

1. 启动本地mysql服务

shell 终端输入：net start mysql80 ，先启动mysql服务

shell 终端输入：mysql -u root -p ，然后输入密码，进入mysql数据库

shell 终端输入：net stop mysql80 ，关闭mysql服务

shell 终端输入：（尽量手打，大小写皆可，注意空格）

create database if not exists 数据库名字 character set utf8mb4; ，创建数据库

其中：show databases; ，查看所有数据库

高危：drop database if exists 数据库名; ，删除某个数据库

2. 安装依赖

conda环境输入：pip install async sqlalchemy ，安装异步 orm 框架，处理数据库交互

conda环境输入：pip install aiomysql ，安装mysql 异步驱动，适配 sqlalchemy 异步引擎

3. 配置异步数据库，db_config.py

from sqlalchemy.ext.asyncio import async_sessionmaker, asyncsession, create_async_engine
# 数据库url
async_database_url = "mysql+aiomysql://root:123456@localhost:3306/pdf_agent?charset=utf8mb4"
# 创建异步引擎
async_engine = create_async_engine(
    async_database_url,  # 传入数据库连接 url
    echo=true,  # 可选：开启 sql 日志输出，便于调试
    pool_size=10,  # 设置连接池中保持的持久连接数为 10
    max_overflow=20  # 设置连接池允许创建的额外连接数最大为 20
)
# 创建异步会话工厂
asyncsessionlocal = async_sessionmaker(
    bind=async_engine,  # 绑定到之前创建的异步引擎
    class_=asyncsession,  # 指定会话类为 asyncsession
    expire_on_commit=false  # 提交后不使对象属性过期，保持数据可访问
)
# 依赖项，用于获取数据库会话
async def get_db():
    async with asyncsessionlocal() as session:
        try:
            yield session  # 将会话对象产出给调用者，暂停函数执行直到请求结束
            await session.commit()  # 如果无异常，提交事务
        except exception:
            await session.rollback()  # 发生异常时回滚事务
            raise  # 重新抛出异常以便上层处理
        finally:
            await session.close()  # 关闭会话，释放资源

4. 创建数据库表

4.1 定义基类

from datetime import datetime, utc
from sqlalchemy import datetime, string, integer
from sqlalchemy.orm import declarativebase, mapped, mapped_column

def utc_now() -> datetime:
    return datetime.now(utc)

class base(declarativebase):
    # 创建时间：默认utc时间，北京时间 = utc + 8 小时
    # utc 是全球统一的标准时间基准，是互联网、服务器、数据库、日志、gps、国际业务通用时间
    created_at: mapped[datetime] = mapped_column(
        datetime,
        default=utc_now,
        comment="创建时间(utc)"
    )
    # 更新时间：新增默认赋值 + 每次update自动刷新
    updated_at: mapped[datetime] = mapped_column(
        datetime,
        default=utc_now,
        onupdate=utc_now,
        comment="更新时间(utc)"
    )

4.2 定义数据库表

from sqlalchemy import integer, string, float

class user(base):
    # 数据库表名
    __tablename__ = "t_user"

    # 字段定义：mapped 标注类型，mapped_column 配置字段属性
    id: mapped[int] = mapped_column(
        integer,
        primary_key=true,
        autoincrement=true,
        name="user_id",
        comment="用户编号"
    )
    name: mapped[str] = mapped_column(
        string(20),
        name="user_name",
        nullable=false,
        comment="用户名"
    )
    password: mapped[str] = mapped_column(
        string(20),
        name="user_password",
        nullable=false,
        comment="密码"
    )

5. 自动创建数据库表

fastapi 提供了事件监听机制，可以在项目启动（startup）时执行指定逻辑，利用该监听机制实现数据库表的自动创建。

from backend.db_models.db_file import base
from backend.config.db_conf import async_engine

# 定义异步建表函数
async def create_tables():
    async with async_engine.begin() as conn:
        # run_sync包裹同步的create_all，创建所有继承base的表）
        await conn.run_sync(base.metadata.create_all)

# fastapi 启动事件：服务运行自动执行建表
@app.on_event("startup")
async def startup_event():
    await create_tables()

启动 fastapi 项目， uvicorn main:app --reload，可以看到表被创建