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，可以看到表被创建
6. PyCharm 手动连接MySQL数据库
依次点击：
输入用户名 root 和 密码，点击测试连接
点击旁边的...，勾上表名 pdf_agent ，可以看到我们建的数据库表，也可以进行操作
