133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""202512151530
|
||
|
||
Revision ID: 2a46f1e9a590
|
||
Revises: 94a98e279951
|
||
Create Date: 2025-12-15 15:27:33.975790
|
||
|
||
"""
|
||
from typing import Sequence, Union
|
||
|
||
from alembic import op
|
||
import sqlalchemy as sa
|
||
from sqlalchemy import inspect
|
||
from sqlalchemy.dialects import postgresql
|
||
|
||
# revision identifiers, used by Alembic.
|
||
revision: str = '2a46f1e9a590'
|
||
down_revision: Union[str, None] = '94a98e279951'
|
||
branch_labels: Union[str, Sequence[str], None] = None
|
||
depends_on: Union[str, Sequence[str], None] = None
|
||
|
||
|
||
def table_exists(table_name: str) -> bool:
|
||
"""检查表是否存在"""
|
||
bind = op.get_bind()
|
||
inspector = inspect(bind)
|
||
return table_name in inspector.get_table_names()
|
||
|
||
|
||
def column_exists(table_name: str, column_name: str) -> bool:
|
||
"""检查列是否存在"""
|
||
bind = op.get_bind()
|
||
inspector = inspect(bind)
|
||
if not table_exists(table_name):
|
||
return False
|
||
columns = [col['name'] for col in inspector.get_columns(table_name)]
|
||
return column_name in columns
|
||
|
||
|
||
def index_exists(table_name: str, index_name: str) -> bool:
|
||
"""检查索引是否存在"""
|
||
bind = op.get_bind()
|
||
inspector = inspect(bind)
|
||
if not table_exists(table_name):
|
||
return False
|
||
indexes = [idx['name'] for idx in inspector.get_indexes(table_name)]
|
||
return index_name in indexes
|
||
|
||
|
||
def constraint_exists(table_name: str, constraint_name: str) -> bool:
|
||
"""检查约束是否存在（外键、唯一约束等）"""
|
||
bind = op.get_bind()
|
||
inspector = inspect(bind)
|
||
if not table_exists(table_name):
|
||
return False
|
||
|
||
# 检查外键约束
|
||
foreign_keys = [fk['name'] for fk in inspector.get_foreign_keys(table_name) if fk['name']]
|
||
if constraint_name in foreign_keys:
|
||
return True
|
||
|
||
# 检查唯一约束
|
||
unique_constraints = [uc['name'] for uc in inspector.get_unique_constraints(table_name) if uc['name']]
|
||
if constraint_name in unique_constraints:
|
||
return True
|
||
|
||
# 检查检查约束
|
||
check_constraints = [cc['name'] for cc in inspector.get_check_constraints(table_name) if cc['name']]
|
||
if constraint_name in check_constraints:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def trigger_exists(trigger_name: str) -> bool:
|
||
"""检查触发器是否存在（PostgreSQL）"""
|
||
bind = op.get_bind()
|
||
result = bind.execute(sa.text(
|
||
"SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = :trigger_name)"
|
||
), {"trigger_name": trigger_name})
|
||
return result.scalar()
|
||
|
||
|
||
def sequence_exists(sequence_name: str) -> bool:
|
||
"""检查序列是否存在（PostgreSQL）"""
|
||
bind = op.get_bind()
|
||
result = bind.execute(sa.text(
|
||
"SELECT EXISTS (SELECT 1 FROM pg_class WHERE relkind = 'S' AND relname = :sequence_name)"
|
||
), {"sequence_name": sequence_name})
|
||
return result.scalar()
|
||
|
||
|
||
def enum_exists(enum_name: str) -> bool:
|
||
"""检查枚举类型是否存在（PostgreSQL）"""
|
||
bind = op.get_bind()
|
||
result = bind.execute(sa.text(
|
||
"SELECT EXISTS (SELECT 1 FROM pg_type WHERE typname = :enum_name)"
|
||
), {"enum_name": enum_name})
|
||
return result.scalar()
|
||
|
||
|
||
def upgrade() -> None:
|
||
# ### commands auto generated by Alembic - please adjust! ###
|
||
with op.batch_alter_table('api_keys', schema=None) as batch_op:
|
||
batch_op.add_column(sa.Column('api_key', sa.String(length=255), nullable=False, comment='API Key 明文'))
|
||
batch_op.alter_column('scopes',
|
||
existing_type=postgresql.JSONB(astext_type=sa.Text()),
|
||
nullable=True,
|
||
existing_comment='权限范围列表')
|
||
batch_op.drop_index(batch_op.f('ix_api_keys_key_hash'))
|
||
batch_op.create_index(batch_op.f('ix_api_keys_api_key'), ['api_key'], unique=True)
|
||
batch_op.drop_column('key_hash')
|
||
batch_op.drop_column('resource_type')
|
||
batch_op.drop_column('key_prefix')
|
||
|
||
# ### end Alembic commands ###
|
||
|
||
|
||
def downgrade() -> None:
|
||
# ### commands auto generated by Alembic - please adjust! ###
|
||
with op.batch_alter_table('api_keys', schema=None) as batch_op:
|
||
batch_op.add_column(sa.Column('key_prefix', sa.VARCHAR(length=20), autoincrement=False, nullable=False, comment='Key 前缀'))
|
||
batch_op.add_column(sa.Column('resource_type', sa.VARCHAR(length=50), autoincrement=False, nullable=True, comment='资源类型'))
|
||
batch_op.add_column(sa.Column('key_hash', sa.VARCHAR(length=255), autoincrement=False, nullable=False, comment='Key 哈希值'))
|
||
batch_op.drop_index(batch_op.f('ix_api_keys_api_key'))
|
||
batch_op.create_index(batch_op.f('ix_api_keys_key_hash'), ['key_hash'], unique=True)
|
||
batch_op.alter_column('scopes',
|
||
existing_type=postgresql.JSONB(astext_type=sa.Text()),
|
||
nullable=False,
|
||
existing_comment='权限范围列表')
|
||
batch_op.drop_column('api_key')
|
||
|
||
# ### end Alembic commands ###
|