INNER CODE UNIT · Python

execute_destroy_sql_scripts

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

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

    console.tip('销毁 SQL 脚本已执行完成')


def run_alembic(*args: str) -> None:
    """执行 alembic 命令"""
    try:
        subprocess.run(['alembic', *args], cwd=BASE_PATH.parent, check=True)
    except subprocess.CalledProcessError as e:

View source record →