FastAPI+SQLAlchemy:仅在数据库创建后执行数据填充函数
解决FastAPI中仅在数据库首次创建时执行数据填充的问题
下面提供几种实用方案,根据你的场景选择即可:
方法1:检查核心业务表是否存在
通过检查某个核心业务表(比如你的城市表)的存在状态,判断数据库是否为首次创建:
from db import models from db.database import engine from sqlalchemy import inspect # 创建数据库结构检查器 inspector = inspect(engine) # 替换为你数据库中实际的核心表名(比如模型定义的__tablename__值) target_table = "cities" if not inspector.has_table(target_table): # 核心表不存在,说明是首次创建数据库,执行建表+数据填充 models.Base.metadata.create_all(engine) initate_cities() else: # 表已存在,仅确保所有表结构都创建完成(可选) models.Base.metadata.create_all(engine)
方法2:用标记表追踪初始化状态
创建一个专门的标记表记录数据库初始化状态,避免误触发填充逻辑:
from db import models from db.database import engine from sqlalchemy import inspect, Table, Column, Integer # 定义用于标记初始化完成的表 init_marker_table = Table( "_db_initialized", models.Base.metadata, Column("id", Integer, primary_key=True), extend_existing=True # 避免重复定义表报错 ) inspector = inspect(engine) if not inspector.has_table("_db_initialized"): # 标记表不存在,执行首次初始化流程 models.Base.metadata.create_all(engine) initate_cities() # 创建标记表并写入初始化记录 with engine.connect() as conn: conn.execute(init_marker_table.insert().values(id=1)) conn.commit() else: # 已完成初始化,仅同步表结构 models.Base.metadata.create_all(engine)
这种方案的优势是:即使手动删除业务表但保留数据库,也不会重复执行填充,只有整个数据库被删除重建时才会触发。
方法3:检查表是否为空(适用于清空表后需重填的场景)
如果需要在核心表被清空时也自动填充数据,可通过检查表行数判断:
from db import models from db.database import engine from sqlalchemy import select, func from db.models import City # 替换为你的核心模型类 models.Base.metadata.create_all(engine) # 检查核心表是否无数据 with engine.connect() as conn: data_count = conn.execute(select(func.count()).select_from(City)).scalar() if data_count == 0: initate_cities() conn.commit()
内容的提问来源于stack exchange,提问作者Tomas Am
相关产品推荐
相关产品推荐

