按业务模块拆分附件权限
This commit is contained in:
Cheng Zhou
@@ -0,0 +1,213 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user