80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
|
"""001_add_information_fields
|
|||
|
|
|
|||
|
|
添加资讯表新字段：包装、评级信息、分类、行情编号
|
|||
|
|
|
|||
|
|
Revision ID: 001
|
|||
|
|
Revises:
|
|||
|
|
Create Date: 2026-04-12
|
|||
|
|
"""
|
|||
|
|
from alembic import op
|
|||
|
|
import sqlalchemy as sa
|
|||
|
|
|
|||
|
|
|
|||
|
|
# revision identifiers, used by Alembic.
|
|||
|
|
revision = '001'
|
|||
|
|
down_revision = None
|
|||
|
|
branch_labels = None
|
|||
|
|
depends_on = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def upgrade() -> None:
|
|||
|
|
"""添加information表的新字段"""
|
|||
|
|
|
|||
|
|
# 检查字段是否存在，如果不存在则添加
|
|||
|
|
# packaging 包装类型
|
|||
|
|
if not _column_exists('information', 'packaging'):
|
|||
|
|
op.add_column('information', sa.Column('packaging', sa.String(50), nullable=True))
|
|||
|
|
|
|||
|
|
# is_graded 是否评级
|
|||
|
|
if not _column_exists('information', 'is_graded'):
|
|||
|
|
op.add_column('information', sa.Column('is_graded', sa.Boolean(), server_default='false', nullable=True))
|
|||
|
|
|
|||
|
|
# grading_company 评级公司
|
|||
|
|
if not _column_exists('information', 'grading_company'):
|
|||
|
|
op.add_column('information', sa.Column('grading_company', sa.String(100), nullable=True))
|
|||
|
|
|
|||
|
|
# grading_score 评级分数
|
|||
|
|
if not _column_exists('information', 'grading_score'):
|
|||
|
|
op.add_column('information', sa.Column('grading_score', sa.String(50), nullable=True))
|
|||
|
|
|
|||
|
|
# category 分类
|
|||
|
|
if not _column_exists('information', 'category'):
|
|||
|
|
op.add_column('information', sa.Column('category', sa.String(100), nullable=True))
|
|||
|
|
|
|||
|
|
# deal_no 行情编号
|
|||
|
|
if not _column_exists('information', 'deal_no'):
|
|||
|
|
op.add_column('information', sa.Column('deal_no', sa.String(50), nullable=True))
|
|||
|
|
op.create_index('ix_information_deal_no', 'information', ['deal_no'])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def downgrade() -> None:
|
|||
|
|
"""删除information表的新字段"""
|
|||
|
|
if _column_exists('information', 'deal_no'):
|
|||
|
|
op.drop_index('ix_information_deal_no', 'information')
|
|||
|
|
op.drop_column('information', 'deal_no')
|
|||
|
|
|
|||
|
|
if _column_exists('information', 'category'):
|
|||
|
|
op.drop_column('information', 'category')
|
|||
|
|
|
|||
|
|
if _column_exists('information', 'grading_score'):
|
|||
|
|
op.drop_column('information', 'grading_score')
|
|||
|
|
|
|||
|
|
if _column_exists('information', 'grading_company'):
|
|||
|
|
op.drop_column('information', 'grading_company')
|
|||
|
|
|
|||
|
|
if _column_exists('information', 'is_graded'):
|
|||
|
|
op.drop_column('information', 'is_graded')
|
|||
|
|
|
|||
|
|
if _column_exists('information', 'packaging'):
|
|||
|
|
op.drop_column('information', 'packaging')
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _column_exists(table: str, column: str) -> bool:
|
|||
|
|
"""检查列是否存在"""
|
|||
|
|
from sqlalchemy import inspect
|
|||
|
|
from app.core.database import engine
|
|||
|
|
|
|||
|
|
inspector = inspect(engine)
|
|||
|
|
columns = [col['name'] for col in inspector.get_columns(table)]
|
|||
|
|
return column in columns
|