35 lines
1.0 KiB
Python
35 lines
1.0 KiB
Python
|
"""Alembic 迁移共享工具（仅用于 versions/ 下的迁移脚本）。"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import sqlalchemy as sa
|
|||
|
|
from alembic import op
|
|||
|
|
|
|||
|
|
|
|||
|
|
def table_exists(table_name: str) -> bool:
|
|||
|
|
bind = op.get_bind()
|
|||
|
|
return table_name in sa.inspect(bind).get_table_names()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def has_column(table_name: str, column_name: str) -> bool:
|
|||
|
|
if not table_exists(table_name):
|
|||
|
|
return False
|
|||
|
|
bind = op.get_bind()
|
|||
|
|
columns = sa.inspect(bind).get_columns(table_name)
|
|||
|
|
return any(column["name"] == column_name for column in columns)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def add_column_if_missing(table_name: str, column: sa.Column) -> bool:
|
|||
|
|
"""若列不存在则 add_column；返回是否执行了添加。"""
|
|||
|
|
if has_column(table_name, column.name):
|
|||
|
|
return False
|
|||
|
|
op.add_column(table_name, column)
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
|
|||
|
|
def drop_column_if_exists(table_name: str, column_name: str) -> bool:
|
|||
|
|
if not has_column(table_name, column_name):
|
|||
|
|
return False
|
|||
|
|
op.drop_column(table_name, column_name)
|
|||
|
|
return True
|