Cheng Zhou
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""rename knowledge note permissions to precautions
|
|
|
|
Revision ID: 20260527_07_permissions
|
|
Revises: 20260527_06
|
|
Create Date: 2026-05-27 16:35:00.000000
|
|
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision: str = "20260527_07_permissions"
|
|
down_revision: Union[str, None] = "20260527_06"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
PERMISSION_KEY_RENAMES = (
|
|
("knowledge_notes:create", "precautions:create"),
|
|
("knowledge_notes:list", "precautions:list"),
|
|
("knowledge_notes:read", "precautions:read"),
|
|
("knowledge_notes:update", "precautions:update"),
|
|
("knowledge_notes:delete", "precautions:delete"),
|
|
("knowledge_notes_attachments:create", "precautions_attachments:create"),
|
|
("knowledge_notes_attachments:read", "precautions_attachments:read"),
|
|
("knowledge_notes_attachments:delete", "precautions_attachments:delete"),
|
|
)
|
|
|
|
|
|
def _table_exists(inspector: sa.Inspector, table_name: str) -> bool:
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def _rename_api_endpoint_permissions(old_key: str, new_key: str) -> None:
|
|
op.execute(
|
|
f"""
|
|
DELETE FROM api_endpoint_permissions AS old_permissions
|
|
WHERE old_permissions.endpoint_key = '{old_key}'
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM api_endpoint_permissions AS new_permissions
|
|
WHERE new_permissions.study_id = old_permissions.study_id
|
|
AND new_permissions.role = old_permissions.role
|
|
AND new_permissions.endpoint_key = '{new_key}'
|
|
)
|
|
"""
|
|
)
|
|
op.execute(
|
|
f"""
|
|
UPDATE api_endpoint_permissions
|
|
SET endpoint_key = '{new_key}'
|
|
WHERE endpoint_key = '{old_key}'
|
|
"""
|
|
)
|
|
|
|
|
|
def _rename_template_permission_key(table_name: str, old_key: str, new_key: str, *, touch_updated_at: bool = False) -> None:
|
|
updated_at_assignment = ", updated_at = NOW()" if touch_updated_at else ""
|
|
op.execute(
|
|
f"""
|
|
UPDATE {table_name}
|
|
SET permissions = (
|
|
SELECT jsonb_object_agg(
|
|
role_key,
|
|
CASE
|
|
WHEN role_permissions ? '{old_key}' AND NOT role_permissions ? '{new_key}'
|
|
THEN (role_permissions - '{old_key}') || jsonb_build_object('{new_key}', role_permissions -> '{old_key}')
|
|
WHEN role_permissions ? '{old_key}'
|
|
THEN role_permissions - '{old_key}'
|
|
ELSE role_permissions
|
|
END
|
|
)
|
|
FROM jsonb_each(permissions::jsonb) AS role_entries(role_key, role_permissions)
|
|
){updated_at_assignment}
|
|
WHERE EXISTS (
|
|
SELECT 1
|
|
FROM jsonb_each(permissions::jsonb) AS role_entries(role_key, role_permissions)
|
|
WHERE role_permissions ? '{old_key}'
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def _rename_permission_keys(mapping: tuple[tuple[str, str], ...]) -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
|
|
if _table_exists(inspector, "api_endpoint_permissions"):
|
|
for old_key, new_key in mapping:
|
|
_rename_api_endpoint_permissions(old_key, new_key)
|
|
|
|
if _table_exists(inspector, "permission_templates"):
|
|
for old_key, new_key in mapping:
|
|
_rename_template_permission_key("permission_templates", old_key, new_key, touch_updated_at=True)
|
|
|
|
if _table_exists(inspector, "permission_template_versions"):
|
|
for old_key, new_key in mapping:
|
|
_rename_template_permission_key("permission_template_versions", old_key, new_key)
|
|
|
|
|
|
def upgrade() -> None:
|
|
_rename_permission_keys(PERMISSION_KEY_RENAMES)
|
|
|
|
|
|
def downgrade() -> None:
|
|
_rename_permission_keys(tuple((new_key, old_key) for old_key, new_key in PERMISSION_KEY_RENAMES))
|