Files
model-platform/migrations/versions/j4k5l6m7n8o9_add_system_role_view_permission.py
2026-08-25 22:33:25 +08:00

120 lines
4.1 KiB
Python

"""add system:role:view permission + grant it to admin
Revision ID: j4k5l6m7n8o9
Revises: e1f2a3b4c5d6
Create Date: 2026-08-25
"""
import hashlib
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "j4k5l6m7n8o9"
down_revision: str | Sequence[str] | None = "e1f2a3b4c5d6"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
# ── seed constants ───────────────────────────────────────────────
ADMIN_ROLE_ID = "0000000000000000000000000A"
PERMISSION_CODE = "system:role:view"
PERMISSION_NAME = "查看角色管理"
MODULE_CODE = "system"
PERMISSION_DESCRIPTION = "查看角色管理与权限配置"
CROCKFORD_BASE32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
def _deterministic_permission_id(code: str) -> str:
"""Stable 26-char ULID-shaped id derived from permission_code.
Mirrors ``e1f2a3b4c5d6_rebuild_baseline._deterministic_permission_id``
so the id for a given code matches what the baseline would derive.
"""
digest = hashlib.sha256(
f"model-platform-permission-v1:{code}".encode()
).digest()
value = int.from_bytes(b"\x00" * 6 + digest[:10], byteorder="big")
encoded = ["0"] * 26
for index in range(25, -1, -1):
encoded[index] = CROCKFORD_BASE32[value & 31]
value >>= 5
return "".join(encoded)
def upgrade() -> None:
"""Add system:role:view permission and grant it to admin only."""
permission_id = _deterministic_permission_id(PERMISSION_CODE)
permissions_table = sa.table(
"permissions",
sa.column("permission_id", sa.CHAR(26)),
sa.column("permission_code", sa.String(128)),
sa.column("permission_name", sa.String(100)),
sa.column("module_code", sa.String(64)),
sa.column("description", sa.String(500)),
)
op.bulk_insert(
permissions_table,
[
{
"permission_id": permission_id,
"permission_code": PERMISSION_CODE,
"permission_name": PERMISSION_NAME,
"module_code": MODULE_CODE,
"description": PERMISSION_DESCRIPTION,
}
],
)
# admin 持有全部 system:*(沿用基线约定);developer 不授予 system:*.
role_permissions_table = sa.table(
"role_permissions",
sa.column("role_id", sa.CHAR(26)),
sa.column("permission_id", sa.CHAR(26)),
)
op.bulk_insert(
role_permissions_table,
[
{
"role_id": ADMIN_ROLE_ID,
"permission_id": permission_id,
}
],
)
def downgrade() -> None:
"""Remove admin's system:role:view grant and soft-delete the permission."""
permission_id = _deterministic_permission_id(PERMISSION_CODE)
# 软删 role_permissions 关联行(与后端 soft-delete 约定一致).
role_permissions_table = sa.table(
"role_permissions",
sa.column("role_id", sa.CHAR(26)),
sa.column("permission_id", sa.CHAR(26)),
sa.column("is_deleted", sa.Integer),
sa.column("deleted_at", sa.DateTime),
)
op.execute(
role_permissions_table.update()
.where(role_permissions_table.c.role_id == ADMIN_ROLE_ID)
.where(role_permissions_table.c.permission_id == permission_id)
.where(role_permissions_table.c.is_deleted == 0)
.values(is_deleted=1, deleted_at=sa.text("CURRENT_TIMESTAMP(3)"))
)
# 软删 permissions 行(基线表含 is_deleted,优先软删除).
permissions_table = sa.table(
"permissions",
sa.column("permission_id", sa.CHAR(26)),
sa.column("is_deleted", sa.Integer),
sa.column("deleted_at", sa.DateTime),
)
op.execute(
permissions_table.update()
.where(permissions_table.c.permission_id == permission_id)
.where(permissions_table.c.is_deleted == 0)
.values(is_deleted=1, deleted_at=sa.text("CURRENT_TIMESTAMP(3)"))
)