99 lines
2.2 KiB
Mako
99 lines
2.2 KiB
Mako
"""${message}
|
||
|
||
Revision ID: ${up_revision}
|
||
Revises: ${down_revision | comma,n}
|
||
Create Date: ${create_date}
|
||
|
||
"""
|
||
from alembic import op
|
||
import sqlalchemy as sa
|
||
from sqlalchemy import text
|
||
from sqlalchemy.engine import reflection
|
||
import logging
|
||
${imports if imports else ""}
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger('alembic.migration')
|
||
|
||
# revision identifiers, used by Alembic.
|
||
revision = ${repr(up_revision)}
|
||
down_revision = ${repr(down_revision)}
|
||
branch_labels = ${repr(branch_labels)}
|
||
depends_on = ${repr(depends_on)}
|
||
|
||
|
||
def safe_execute(func, *args, **kwargs):
|
||
"""
|
||
安全执行函数，捕获异常并记录日志
|
||
|
||
Args:
|
||
func: 要执行的函数
|
||
*args: 位置参数
|
||
**kwargs: 关键字参数
|
||
|
||
Returns:
|
||
执行结果或 None（如果发生异常）
|
||
"""
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except Exception as e:
|
||
logger.error(f"执行 {func.__name__} 时发生错误: {e}")
|
||
return None
|
||
|
||
|
||
def column_exists(table_name, column_name):
|
||
"""
|
||
检查列是否存在
|
||
|
||
Args:
|
||
table_name: 表名
|
||
column_name: 列名
|
||
|
||
Returns:
|
||
bool: 列是否存在
|
||
"""
|
||
conn = op.get_bind()
|
||
inspector = reflection.Inspector.from_engine(conn)
|
||
columns = inspector.get_columns(table_name)
|
||
return column_name in [column['name'] for column in columns]
|
||
|
||
|
||
def index_exists(table_name, index_name):
|
||
"""
|
||
检查索引是否存在
|
||
|
||
Args:
|
||
table_name: 表名
|
||
index_name: 索引名
|
||
|
||
Returns:
|
||
bool: 索引是否存在
|
||
"""
|
||
conn = op.get_bind()
|
||
inspector = reflection.Inspector.from_engine(conn)
|
||
indexes = inspector.get_indexes(table_name)
|
||
return index_name in [index['name'] for index in indexes]
|
||
|
||
|
||
def table_exists(table_name):
|
||
"""
|
||
检查表是否存在
|
||
|
||
Args:
|
||
table_name: 表名
|
||
|
||
Returns:
|
||
bool: 表是否存在
|
||
"""
|
||
conn = op.get_bind()
|
||
inspector = reflection.Inspector.from_engine(conn)
|
||
return table_name in inspector.get_table_names()
|
||
|
||
|
||
def upgrade():
|
||
${upgrades if upgrades else "pass"}
|
||
|
||
|
||
def downgrade():
|
||
${downgrades if downgrades else "pass"} |