65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
|
"""撤下配置发布的"审核人不得等于创建人"检查约束。
|
|||
|
|
|
|||
|
|
为什么撤下：单管理员部署下自审是唯一可行的发布路径。原约束
|
|||
|
|
`chk_config_release_separation`（`reviewer_id IS NULL OR reviewer_id <> created_by`）
|
|||
|
|
让"创建人自审"在数据库层被拒绝，表现为审核接口 500 而不是业务异常；同时自审若按
|
|||
|
|
"留空 reviewer_id"规避约束，又会被 `admin_service` 的激活分支判为"版本未审核"
|
|||
|
|
（该分支要求 `reviewer_id is not None`），自审版本因此无法激活。
|
|||
|
|
|
|||
|
|
服务层 `ConfigReleaseService.approve`、`AdminService` 均已按"允许自审、如实写入
|
|||
|
|
reviewer_id"实现，本迁移补齐缺失的数据库侧变更。
|
|||
|
|
|
|||
|
|
对基线的影响：本迁移只删除 `config_release` 上的一个 CHECK 约束，不新增、不重命名、
|
|||
|
|
不删除任何表或字段，不改变任何字段的类型、可空性与业务含义；`chk_config_release_status`
|
|||
|
|
约束以及全部外键、唯一键、索引保持原样（可用 `tools/schema_fingerprint.py` 与
|
|||
|
|
`tools/audit_constraints.py` 复核）。
|
|||
|
|
|
|||
|
|
回滚：`downgrade` 重建该约束。若库中已存在 `reviewer_id = created_by` 的行，重建会被
|
|||
|
|
MySQL 拒绝——这是有意为之：回滚到"双人复核"语义前必须先处理这些自审记录。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
from alembic import op
|
|||
|
|
|
|||
|
|
revision = "20260911_drop_review_separation"
|
|||
|
|
down_revision = "20260911_field_correction"
|
|||
|
|
branch_labels = None
|
|||
|
|
depends_on = None
|
|||
|
|
|
|||
|
|
TABLE = "config_release"
|
|||
|
|
CONSTRAINT = "chk_config_release_separation"
|
|||
|
|
CLAUSE = "reviewer_id IS NULL OR reviewer_id <> created_by"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _constraint_exists() -> bool:
|
|||
|
|
connection = op.get_bind()
|
|||
|
|
found = connection.scalar(
|
|||
|
|
text(
|
|||
|
|
"""
|
|||
|
|
SELECT COUNT(*)
|
|||
|
|
FROM information_schema.TABLE_CONSTRAINTS
|
|||
|
|
WHERE CONSTRAINT_SCHEMA = database()
|
|||
|
|
AND TABLE_NAME = :table
|
|||
|
|
AND CONSTRAINT_NAME = :constraint
|
|||
|
|
AND CONSTRAINT_TYPE = 'CHECK'
|
|||
|
|
"""
|
|||
|
|
),
|
|||
|
|
{"table": TABLE, "constraint": CONSTRAINT},
|
|||
|
|
)
|
|||
|
|
return int(found or 0) > 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def upgrade() -> None:
|
|||
|
|
"""撤下检查约束；约束已不存在时跳过（兼容手工处理过的库）。"""
|
|||
|
|
if not _constraint_exists():
|
|||
|
|
return
|
|||
|
|
op.execute(f"ALTER TABLE {TABLE} DROP CHECK {CONSTRAINT}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def downgrade() -> None:
|
|||
|
|
"""恢复双人复核约束；存在自审记录时 MySQL 会拒绝执行。"""
|
|||
|
|
if _constraint_exists():
|
|||
|
|
return
|
|||
|
|
op.execute(f"ALTER TABLE {TABLE} ADD CONSTRAINT {CONSTRAINT} CHECK ({CLAUSE})")
|