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

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()  # 关闭会话，释放资源

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)"
    )

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="密码"
    )

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()