一、定义Base基类 ORM模型1. 文件划分单独新建db.py文件存放engine、AsyncSessionFactory、Base。❗不要把Base分散写在各个model文件全部从db.py导入Base。db.pyfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 数据库连接字符串 DB_URI mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4 # 1. 创建异步引擎 engine create_async_engine( DB_URI, echoTrue, pool_size10, max_overflow20, pool_timeout10, pool_recycle3600, pool_pre_pingTrue ) # 2. 创建异步会话工厂 AsyncSessionFactory sessionmaker( bindengine, class_AsyncSession, autoflushTrue, expire_on_commitFalse ) # 3. 生成ORM模型父类Base所有数据表模型继承这个Base Base declarative_base()2. 创建模型文件 models/users.py导入顺序从db.py导入Base再写数据表模型from sqlalchemy import Column, Integer, String from db import Base # ✅关键继承db.py中的Base class User(Base): __tablename__ user #数据库真实表名 id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue) username Column(String(100), uniqueTrue) password Column(String(200), nullableFalse)项目目录结构07sqlalchemy/ ├── db.py # engine、SessionFactory、Base在这里 ├── models/ │ ├── __init__.py │ └── users.py # User模型from db import Base ├── alembic/ # alembic迁移仓库init生成 │ ├── versions/ │ ├── env.py # 需要修改 │ ├── script.py.mako │ └── README ├── alembic.ini # 需要修改 └── main.py导入关系db.py生成Base对象models/users.pyfrom db import Base模型类继承Basealembic/env.py需要导入db.py中的BaseAlembic靠Base.metadata扫描所有ORM模型如果env.py没有正确导入Base自动迁移识别不到模型表二、Alembic数据库迁移Alembic类似git管理数据库表版本变更可以升级、回滚表结构模型改动不能直接操作数据库走alembic迁移。1. 安装alembicpip install alembic1.13.22. 创建异步迁移仓库FastAPISQLAlchemy异步必须加--template async生成适配异步mysql的模板alembic init alembic --template async执行完项目根目录生成alembic文件夹 alembic.ini配置文件。2.1 修改文件①alembic.ini找到文件中sqlalchemy.url修改为异步mysql连接串sqlalchemy.url mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4注意ini文件不要加引号直接写连接字符串2.2 修改文件②alembic/env.py默认env.py不会识别你的ORM模型需要两件事把项目根目录加入python路径让env.py能import到db.py导入db.py的Base赋值给target_metadata修改alembic/env.py顶部import sys from os.path import abspath, dirname # 将项目根目录加到导入路径解决找不到db模块报错 sys.path.insert(0, dirname(dirname(abspath(__file__)))) from db import Base # 导入db里面的Base对象 # 修改这一行告诉alembic读取Base下全部模型metadata target_metadata Base.metadata踩坑提醒如果不把项目根目录加入sys.path运行alembic会报ModuleNotFoundError: No module named db如果target_metadata Base.metadata写错--autogenerate检测不到模型变化生成空迁移脚本。models下所有模型只要继承db.Base就会被Base.metadata收集不需要在env.py逐个import每个model但要保证程序加载过model文件为保险也可以在db.py显式导入models。可选在db.py末尾导入模型保证metadata收集完整# db.py末尾追加保证所有模型被加载 from models.users import User2.3 生成迁移脚本模型发生新增/修改字段后执行alembic revision --autogenerate -m 创建user表-m填写本次变更备注--autogenerate对比模型和数据库现状自动生成迁移代码脚本输出目录alembic/versions/下面生成.py迁移脚本autogenerate只会检测ORM模型定义不会自动处理部分删除字段需要检查生成脚本内容。2.4 执行迁移同步到数据库生成脚本只是生成代码还没有真正执行SQL操作数据库执行升级命令把脚本应用到数据库alembic upgrade headhead代表最新版本数据库会创建表同时生成alembic_version版本记录表。常用补充命令alembic history #查看全部迁移版本记录 alembic downgrade -1 #回滚上一个版本 alembic current #查看当前数据库版本三、完整流程梳理导入顺序记忆db.py创建engine、session工厂实例化Base declarative_base()**models/**下各个模型from db import Base定义数据表类继承Basedb.py末尾导入所有model确保Base.metadata收集全部表结构alembic init alembic --template async初始化异步迁移仓库修改alembic.ini填入异步数据库连接url修改alembic/env.py添加项目路径from db import Basetarget_metadata Base.metadataalembic revision --autogenerate -m 备注生成迁移脚本alembic upgrade head真正执行SQL在数据库生成表小结Base一定要统一放在db.py所有模型从db导入Base不要每个模型文件单独生成declarative_base()否则metadata隔离迁移识别不到。Alembic异步项目初始化必须带上--template async。env.py要解决模块导入路径问题target_metadata赋值Base.metadata是核心否则自动生成脚本为空。revision只生成脚本必须执行upgrade head才会改动数据库。FastAPI SQLAlchemy 异步 MySQL 项目结构整理Alembic 迁移改造说明新建models目录拆分__init__.py、users.py模型文件修改 alembic 的env.py删除旧的单文件db.py、旧model.py。 依赖sqlalchemy[asyncio]、asyncmy、alembic项目目录结构your_project/ ├── alembic/ # alembic迁移目录 ├── models/ # 模型包新建 │ ├── __init__.py # db引擎、session、Base对象 │ └── users.py # User表模型 ├── alembic.ini └── main.py1. models/init.py原 db.py 迁移到此from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 异步mysql连接地址 DB_URI mysqlasyncmy://root:123456127.0.0.1:3306/music_db?charsetutf8mb4 # 创建异步引擎 engine create_async_engine( DB_URI, echoTrue, # 打印SQL日志 pool_size10, # 连接池大小 max_overflow20, # 最大溢出连接 pool_timeout10, # 获取连接超时 pool_recycle3600, # 连接回收时长 pool_pre_pingTrue, # 使用前预检测连接有效性 ) # 异步session工厂 AsyncSessionFactory sessionmaker( bindengine, class_AsyncSession, autoflushTrue, expire_on_commitFalse ) # 模型基类所有数据表模型继承该类 Base declarative_base() # 导入模型保证alembic autogenerate可以检测到表模型 from . import users2. models/users.py 用户模型原 model.py 拆分from sqlalchemy import Column, Integer, String from . import Base class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, autoincrementTrue, indexTrue) email Column(String(100), uniqueTrue, indexTrue) username Column(String(100), uniqueTrue, indexTrue) password Column(String(100))3. alembic/env.py 修改后的完整代码import asyncio from logging.config import fileConfig from sqlalchemy import pool from sqlalchemy.engine import Connection from sqlalchemy.ext.asyncio import async_engine_from_config from alembic import context config context.config if config.config_file_name is not None: fileConfig(config.config_file_name) # 关键修改导入models包的Base识别所有数据表元数据 from models import Base target_metadata Base.metadata def run_migrations_offline() - None: url config.get_main_option(sqlalchemy.url) context.configure( urlurl, target_metadatatarget_metadata, literal_bindsTrue, dialect_opts{paramstyle: named}, ) with context.begin_transaction(): context.run_migrations() def do_run_migrations(connection: Connection) - None: context.configure(connectionconnection, target_metadatatarget_metadata) with context.begin_transaction(): context.run_migrations() async def run_async_migrations() - None: connectable async_engine_from_config( config.get_section(config.config_ini_section, {}), prefixsqlalchemy., poolclasspool.NullPool, ) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) await connectable.dispose() def run_migrations_online() - None: asyncio.run(run_async_migrations()) if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()4. alembic.ini 配置片段sqlalchemy.url mysqlasyncmy://root:123456127.0.0.1:3306/music_db?charsetutf8mb4改造要点说明目录重构建立models目录作为数据模型包不再把数据库引擎、模型全部堆在根目录。models/__init__.py存放数据库连接引擎engine、异步会话工厂AsyncSessionFactory、ORM 基类Base末尾导入各业务模型文件保证 Alembic 可以扫描识别数据表。models/users.py存放 User 数据表实体模型继承来自models.__init__.py的Base。修改 alembic/env.py导入from models import Base给target_metadata赋值alembic autogenerate才能自动识别模型变更生成迁移脚本。清理旧文件删除项目根目录下旧的db.py、model.py所有数据库相关代码迁移至 models 包。迁移命令alembic init alembic alembic revision --autogenerate -m init user table alembic upgrade head FastAPI+SQLAlchemy2.0 异步 ORM 模型与 Alembic 迁移
一、定义Base基类 ORM模型1. 文件划分单独新建db.py文件存放engine、AsyncSessionFactory、Base。❗不要把Base分散写在各个model文件全部从db.py导入Base。db.pyfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 数据库连接字符串 DB_URI mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4 # 1. 创建异步引擎 engine create_async_engine( DB_URI, echoTrue, pool_size10, max_overflow20, pool_timeout10, pool_recycle3600, pool_pre_pingTrue ) # 2. 创建异步会话工厂 AsyncSessionFactory sessionmaker( bindengine, class_AsyncSession, autoflushTrue, expire_on_commitFalse ) # 3. 生成ORM模型父类Base所有数据表模型继承这个Base Base declarative_base()2. 创建模型文件 models/users.py导入顺序从db.py导入Base再写数据表模型from sqlalchemy import Column, Integer, String from db import Base # ✅关键继承db.py中的Base class User(Base): __tablename__ user #数据库真实表名 id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue) username Column(String(100), uniqueTrue) password Column(String(200), nullableFalse)项目目录结构07sqlalchemy/ ├── db.py # engine、SessionFactory、Base在这里 ├── models/ │ ├── __init__.py │ └── users.py # User模型from db import Base ├── alembic/ # alembic迁移仓库init生成 │ ├── versions/ │ ├── env.py # 需要修改 │ ├── script.py.mako │ └── README ├── alembic.ini # 需要修改 └── main.py导入关系db.py生成Base对象models/users.pyfrom db import Base模型类继承Basealembic/env.py需要导入db.py中的BaseAlembic靠Base.metadata扫描所有ORM模型如果env.py没有正确导入Base自动迁移识别不到模型表二、Alembic数据库迁移Alembic类似git管理数据库表版本变更可以升级、回滚表结构模型改动不能直接操作数据库走alembic迁移。1. 安装alembicpip install alembic1.13.22. 创建异步迁移仓库FastAPISQLAlchemy异步必须加--template async生成适配异步mysql的模板alembic init alembic --template async执行完项目根目录生成alembic文件夹 alembic.ini配置文件。2.1 修改文件①alembic.ini找到文件中sqlalchemy.url修改为异步mysql连接串sqlalchemy.url mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4注意ini文件不要加引号直接写连接字符串2.2 修改文件②alembic/env.py默认env.py不会识别你的ORM模型需要两件事把项目根目录加入python路径让env.py能import到db.py导入db.py的Base赋值给target_metadata修改alembic/env.py顶部import sys from os.path import abspath, dirname # 将项目根目录加到导入路径解决找不到db模块报错 sys.path.insert(0, dirname(dirname(abspath(__file__)))) from db import Base # 导入db里面的Base对象 # 修改这一行告诉alembic读取Base下全部模型metadata target_metadata Base.metadata踩坑提醒如果不把项目根目录加入sys.path运行alembic会报ModuleNotFoundError: No module named db如果target_metadata Base.metadata写错--autogenerate检测不到模型变化生成空迁移脚本。models下所有模型只要继承db.Base就会被Base.metadata收集不需要在env.py逐个import每个model但要保证程序加载过model文件为保险也可以在db.py显式导入models。可选在db.py末尾导入模型保证metadata收集完整# db.py末尾追加保证所有模型被加载 from models.users import User2.3 生成迁移脚本模型发生新增/修改字段后执行alembic revision --autogenerate -m 创建user表-m填写本次变更备注--autogenerate对比模型和数据库现状自动生成迁移代码脚本输出目录alembic/versions/下面生成.py迁移脚本autogenerate只会检测ORM模型定义不会自动处理部分删除字段需要检查生成脚本内容。2.4 执行迁移同步到数据库生成脚本只是生成代码还没有真正执行SQL操作数据库执行升级命令把脚本应用到数据库alembic upgrade headhead代表最新版本数据库会创建表同时生成alembic_version版本记录表。常用补充命令alembic history #查看全部迁移版本记录 alembic downgrade -1 #回滚上一个版本 alembic current #查看当前数据库版本三、完整流程梳理导入顺序记忆db.py创建engine、session工厂实例化Base declarative_base()**models/**下各个模型from db import Base定义数据表类继承Basedb.py末尾导入所有model确保Base.metadata收集全部表结构alembic init alembic --template async初始化异步迁移仓库修改alembic.ini填入异步数据库连接url修改alembic/env.py添加项目路径from db import Basetarget_metadata Base.metadataalembic revision --autogenerate -m 备注生成迁移脚本alembic upgrade head真正执行SQL在数据库生成表小结Base一定要统一放在db.py所有模型从db导入Base不要每个模型文件单独生成declarative_base()否则metadata隔离迁移识别不到。Alembic异步项目初始化必须带上--template async。env.py要解决模块导入路径问题target_metadata赋值Base.metadata是核心否则自动生成脚本为空。revision只生成脚本必须执行upgrade head才会改动数据库。FastAPI SQLAlchemy 异步 MySQL 项目结构整理Alembic 迁移改造说明新建models目录拆分__init__.py、users.py模型文件修改 alembic 的env.py删除旧的单文件db.py、旧model.py。 依赖sqlalchemy[asyncio]、asyncmy、alembic项目目录结构your_project/ ├── alembic/ # alembic迁移目录 ├── models/ # 模型包新建 │ ├── __init__.py # db引擎、session、Base对象 │ └── users.py # User表模型 ├── alembic.ini └── main.py1. models/init.py原 db.py 迁移到此from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 异步mysql连接地址 DB_URI mysqlasyncmy://root:123456127.0.0.1:3306/music_db?charsetutf8mb4 # 创建异步引擎 engine create_async_engine( DB_URI, echoTrue, # 打印SQL日志 pool_size10, # 连接池大小 max_overflow20, # 最大溢出连接 pool_timeout10, # 获取连接超时 pool_recycle3600, # 连接回收时长 pool_pre_pingTrue, # 使用前预检测连接有效性 ) # 异步session工厂 AsyncSessionFactory sessionmaker( bindengine, class_AsyncSession, autoflushTrue, expire_on_commitFalse ) # 模型基类所有数据表模型继承该类 Base declarative_base() # 导入模型保证alembic autogenerate可以检测到表模型 from . import users2. models/users.py 用户模型原 model.py 拆分from sqlalchemy import Column, Integer, String from . import Base class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, autoincrementTrue, indexTrue) email Column(String(100), uniqueTrue, indexTrue) username Column(String(100), uniqueTrue, indexTrue) password Column(String(100))3. alembic/env.py 修改后的完整代码import asyncio from logging.config import fileConfig from sqlalchemy import pool from sqlalchemy.engine import Connection from sqlalchemy.ext.asyncio import async_engine_from_config from alembic import context config context.config if config.config_file_name is not None: fileConfig(config.config_file_name) # 关键修改导入models包的Base识别所有数据表元数据 from models import Base target_metadata Base.metadata def run_migrations_offline() - None: url config.get_main_option(sqlalchemy.url) context.configure( urlurl, target_metadatatarget_metadata, literal_bindsTrue, dialect_opts{paramstyle: named}, ) with context.begin_transaction(): context.run_migrations() def do_run_migrations(connection: Connection) - None: context.configure(connectionconnection, target_metadatatarget_metadata) with context.begin_transaction(): context.run_migrations() async def run_async_migrations() - None: connectable async_engine_from_config( config.get_section(config.config_ini_section, {}), prefixsqlalchemy., poolclasspool.NullPool, ) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) await connectable.dispose() def run_migrations_online() - None: asyncio.run(run_async_migrations()) if context.is_offline_mode(): run_migrations_offline() else: run_migrations_online()4. alembic.ini 配置片段sqlalchemy.url mysqlasyncmy://root:123456127.0.0.1:3306/music_db?charsetutf8mb4改造要点说明目录重构建立models目录作为数据模型包不再把数据库引擎、模型全部堆在根目录。models/__init__.py存放数据库连接引擎engine、异步会话工厂AsyncSessionFactory、ORM 基类Base末尾导入各业务模型文件保证 Alembic 可以扫描识别数据表。models/users.py存放 User 数据表实体模型继承来自models.__init__.py的Base。修改 alembic/env.py导入from models import Base给target_metadata赋值alembic autogenerate才能自动识别模型变更生成迁移脚本。清理旧文件删除项目根目录下旧的db.py、model.py所有数据库相关代码迁移至 models 包。迁移命令alembic init alembic alembic revision --autogenerate -m init user table alembic upgrade head