Cheng Zhou
214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
"""remove generic attachment permissions
|
|
|
|
Revision ID: 20260527_08
|
|
Revises: 20260527_07_permissions
|
|
Create Date: 2026-05-27 17:10:00.000000
|
|
|
|
"""
|
|
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision: str = "20260527_08"
|
|
down_revision: Union[str, None] = "20260527_07_permissions"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
GENERIC_TO_MODULE_KEYS = {
|
|
"attachments:create": (
|
|
"fees_contracts_attachments:create",
|
|
"startup_initiation_attachments:create",
|
|
"startup_ethics_attachments:create",
|
|
"startup_auth_attachments:create",
|
|
"drug_shipments_attachments:create",
|
|
"precautions_attachments:create",
|
|
"faq_attachments:create",
|
|
),
|
|
"attachments:read": (
|
|
"fees_contracts_attachments:read",
|
|
"startup_initiation_attachments:read",
|
|
"startup_ethics_attachments:read",
|
|
"startup_auth_attachments:read",
|
|
"drug_shipments_attachments:read",
|
|
"precautions_attachments:read",
|
|
"faq_attachments:read",
|
|
),
|
|
"attachments:delete": (
|
|
"fees_contracts_attachments:delete",
|
|
"startup_initiation_attachments:delete",
|
|
"startup_ethics_attachments:delete",
|
|
"startup_auth_attachments:delete",
|
|
"drug_shipments_attachments:delete",
|
|
"precautions_attachments:delete",
|
|
"faq_attachments:delete",
|
|
),
|
|
}
|
|
|
|
|
|
def _table_exists(inspector: sa.Inspector, table_name: str) -> bool:
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def _expand_api_endpoint_permission(old_key: str, new_key: str) -> None:
|
|
op.execute(
|
|
f"""
|
|
INSERT INTO api_endpoint_permissions (id, study_id, role, endpoint_key, allowed, updated_at)
|
|
SELECT (md5(random()::text || clock_timestamp()::text))::uuid, source.study_id, source.role, '{new_key}', source.allowed, NOW()
|
|
FROM api_endpoint_permissions AS source
|
|
WHERE source.endpoint_key = '{old_key}'
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM api_endpoint_permissions AS target
|
|
WHERE target.study_id = source.study_id
|
|
AND target.role = source.role
|
|
AND target.endpoint_key = '{new_key}'
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def _remove_api_endpoint_permissions(keys: tuple[str, ...]) -> None:
|
|
quoted_keys = ", ".join(f"'{key}'" for key in keys)
|
|
op.execute(f"DELETE FROM api_endpoint_permissions WHERE endpoint_key IN ({quoted_keys})")
|
|
|
|
|
|
def _expand_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 || jsonb_build_object('{new_key}', 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 _remove_template_permission_key(table_name: str, 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, role_permissions - '{key}')
|
|
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 ? '{key}'
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def _restore_api_endpoint_permission(old_key: str, new_keys: tuple[str, ...]) -> None:
|
|
quoted_new_keys = ", ".join(f"'{key}'" for key in new_keys)
|
|
op.execute(
|
|
f"""
|
|
INSERT INTO api_endpoint_permissions (id, study_id, role, endpoint_key, allowed, updated_at)
|
|
SELECT (md5(random()::text || clock_timestamp()::text))::uuid, source.study_id, source.role, '{old_key}', bool_or(source.allowed), NOW()
|
|
FROM api_endpoint_permissions AS source
|
|
WHERE source.endpoint_key IN ({quoted_new_keys})
|
|
GROUP BY source.study_id, source.role
|
|
HAVING NOT EXISTS (
|
|
SELECT 1
|
|
FROM api_endpoint_permissions AS target
|
|
WHERE target.study_id = source.study_id
|
|
AND target.role = source.role
|
|
AND target.endpoint_key = '{old_key}'
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def _restore_template_permission_key(table_name: str, old_key: str, new_keys: tuple[str, ...], *, touch_updated_at: bool = False) -> None:
|
|
updated_at_assignment = ", updated_at = NOW()" if touch_updated_at else ""
|
|
new_key_checks = " OR ".join(f"role_permissions ? '{key}'" for key in new_keys)
|
|
new_key_values = ", ".join(f"role_permissions -> '{key}'" for key in new_keys)
|
|
op.execute(
|
|
f"""
|
|
UPDATE {table_name}
|
|
SET permissions = (
|
|
SELECT jsonb_object_agg(
|
|
role_key,
|
|
CASE
|
|
WHEN NOT role_permissions ? '{old_key}' AND ({new_key_checks})
|
|
THEN role_permissions || jsonb_build_object('{old_key}', COALESCE({new_key_values}))
|
|
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 {new_key_checks}
|
|
)
|
|
"""
|
|
)
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
old_keys = tuple(GENERIC_TO_MODULE_KEYS)
|
|
|
|
if _table_exists(inspector, "api_endpoint_permissions"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
for new_key in new_keys:
|
|
_expand_api_endpoint_permission(old_key, new_key)
|
|
_remove_api_endpoint_permissions(old_keys)
|
|
|
|
if _table_exists(inspector, "permission_templates"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
for new_key in new_keys:
|
|
_expand_template_permission_key("permission_templates", old_key, new_key, touch_updated_at=True)
|
|
_remove_template_permission_key("permission_templates", old_key, touch_updated_at=True)
|
|
|
|
if _table_exists(inspector, "permission_template_versions"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
for new_key in new_keys:
|
|
_expand_template_permission_key("permission_template_versions", old_key, new_key)
|
|
_remove_template_permission_key("permission_template_versions", old_key)
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
module_keys = tuple(key for new_keys in GENERIC_TO_MODULE_KEYS.values() for key in new_keys)
|
|
|
|
if _table_exists(inspector, "api_endpoint_permissions"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
_restore_api_endpoint_permission(old_key, new_keys)
|
|
_remove_api_endpoint_permissions(module_keys)
|
|
|
|
if _table_exists(inspector, "permission_templates"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
_restore_template_permission_key("permission_templates", old_key, new_keys, touch_updated_at=True)
|
|
for new_key in new_keys:
|
|
_remove_template_permission_key("permission_templates", new_key, touch_updated_at=True)
|
|
|
|
if _table_exists(inspector, "permission_template_versions"):
|
|
for old_key, new_keys in GENERIC_TO_MODULE_KEYS.items():
|
|
_restore_template_permission_key("permission_template_versions", old_key, new_keys)
|
|
for new_key in new_keys:
|
|
_remove_template_permission_key("permission_template_versions", new_key)
|