INNER CODE UNIT · Python

execute_sql_scripts

fastapi-practices/fba-slim · backend/cli.py:529

async def execute_sql_scripts(db: AsyncSession, sql_scripts: str, *, is_init: bool = False) -> None:
    """解析并执行 SQL 脚本"""
    try:
        stmts = await parse_sql_script(sql_scripts)
        conn = await db.connection()
        for stmt in stmts:
            await conn.exec_driver_sql(stmt)
    except Exception as e:
        raise cappa.Exit(f'SQL 脚本执行失败：{e}', code=1)

    if not is_init:
        console.tip('SQL 脚本已执行完成')


async def execute_destroy_sql_scripts(db: AsyncSession, sql_scripts: str) -> None:
    """执行插件销毁 SQL 脚本"""
    try:
        stmts = await parse_sql_script(sql_scripts, is_destroy=True)

View source record →