FastAPI连接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.pyfrom sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession, create_async_engine # 数据库URL ASYNC_DATABASE_URL mysqlaiomysql://root:123456localhost:3306/pdf_agent?charsetutf8mb4 # 创建异步引擎 async_engine create_async_engine( ASYNC_DATABASE_URL, # 传入数据库连接 URL echoTrue, # 可选开启 SQL 日志输出便于调试 pool_size10, # 设置连接池中保持的持久连接数为 10 max_overflow20 # 设置连接池允许创建的额外连接数最大为 20 ) # 创建异步会话工厂 AsyncSessionLocal async_sessionmaker( bindasync_engine, # 绑定到之前创建的异步引擎 class_AsyncSession, # 指定会话类为 AsyncSession expire_on_commitFalse # 提交后不使对象属性过期保持数据可访问 ) # 依赖项用于获取数据库会话 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, defaultutc_now, comment创建时间(UTC) ) # 更新时间新增默认赋值 每次update自动刷新 updated_at: Mapped[datetime] mapped_column( DateTime, defaultutc_now, onupdateutc_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_keyTrue, autoincrementTrue, nameuser_id, comment用户编号 ) name: Mapped[str] mapped_column( String(20), nameuser_name, nullableFalse, comment用户名 ) password: Mapped[str] mapped_column( String(20), nameuser_password, nullableFalse, 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 可以看到我们建的数据库表也可以进行操作