| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081 |
- """用户界面主题偏好:sys_user 加 ui_theme 列(契约1.20)。
- Revision ID: 0027_ui_theme
- Revises: 0026_borrow_permission
- Create Date: 2026-09-03
- 变更要点:
- 1. 主题为账号级漫游偏好:任意设备登录后跟随账号;blue 为缺省
- (存量用户与未设置路径统一回退蓝色);
- 2. CHECK 白名单 blue/red/green(convention 渲染名
- ck_sys_user_user_ui_theme,与模型 CheckConstraint(name="user_ui_theme") 一致);
- 3. 达梦兼容:探测统一走 SQLAlchemy inspector、约束名大小写不敏感比较,
- 删除以库中实际名执行(达梦渲染为大写);
- 4. 升级幂等:列/约束先探测,与 Flyway V12 并行执行互不冲突;
- 降级删列(偏好可丢失,无审计意义)。
- """
- from __future__ import annotations
- from collections.abc import Sequence
- import sqlalchemy as sa
- from alembic import op
- revision: str = "0027_ui_theme"
- down_revision: str | None = "0026_borrow_permission"
- branch_labels: str | Sequence[str] | None = None
- depends_on: str | Sequence[str] | None = None
- _TABLE = "sys_user"
- _COLUMN = "ui_theme"
- # naming convention 渲染后的最终约束名(ck_%(table)s_%(constraint_name)s)
- _CHECK_NAME = "ck_sys_user_user_ui_theme"
- def _columns() -> set[str]:
- inspector = sa.inspect(op.get_bind())
- return {column["name"] for column in inspector.get_columns(_TABLE)}
- def _check_exists() -> bool:
- existing = {
- item["name"].lower()
- for item in sa.inspect(op.get_bind()).get_check_constraints(_TABLE)
- }
- return _CHECK_NAME in existing
- def _drop_check() -> None:
- # 达梦约束名为大写,以库中实际名执行删除。
- for item in sa.inspect(op.get_bind()).get_check_constraints(_TABLE):
- if item["name"].lower() == _CHECK_NAME:
- op.get_bind().execute(
- sa.text(f"ALTER TABLE {_TABLE} DROP CONSTRAINT {item['name']}")
- )
- return
- def upgrade() -> None:
- if _COLUMN not in _columns():
- op.add_column(
- _TABLE,
- sa.Column(
- _COLUMN,
- sa.String(length=12),
- nullable=False,
- server_default=sa.text("'blue'"),
- ),
- )
- if not _check_exists():
- op.create_check_constraint(
- "user_ui_theme",
- _TABLE,
- "ui_theme IN ('blue', 'red', 'green')",
- )
- def downgrade() -> None:
- _drop_check()
- if _COLUMN in _columns():
- op.drop_column(_TABLE, _COLUMN)
|