refactor: bulk updating Admin to Editor roles

- Consolidated the migration process for search space memberships and invites from Admin to Editor roles using bulk SQL updates.
- Removed the Admin role in bulk for system roles.
- Updated permissions for Editor and Viewer roles across all search spaces in a more efficient manner.
This commit is contained in:
DESKTOP-RTLN3BA\$punk 2026-01-20 03:11:49 -08:00
parent e578bb9e77
commit f200502ffc

View file

@ -67,63 +67,42 @@ NEW_VIEWER_PERMISSIONS = [
def upgrade(): def upgrade():
connection = op.get_bind() connection = op.get_bind()
# Step 1: For each search space, get the Editor role ID and Admin role ID # Step 1: Move all memberships from Admin roles to corresponding Editor roles (BULK)
search_spaces = connection.execute( # Uses a subquery to match Admin->Editor within the same search space
sa.text("SELECT id FROM searchspaces") connection.execute(
).fetchall() sa.text("""
UPDATE search_space_memberships m
SET role_id = e.id
FROM search_space_roles a
JOIN search_space_roles e ON a.search_space_id = e.search_space_id
WHERE m.role_id = a.id
AND a.name = 'Admin'
AND e.name = 'Editor'
""")
)
for (ss_id,) in search_spaces: # Step 2: Move all invites from Admin roles to corresponding Editor roles (BULK)
# Get Admin and Editor role IDs for this search space connection.execute(
admin_role = connection.execute( sa.text("""
sa.text(""" UPDATE search_space_invites i
SELECT id FROM search_space_roles SET role_id = e.id
WHERE search_space_id = :ss_id AND name = 'Admin' FROM search_space_roles a
"""), JOIN search_space_roles e ON a.search_space_id = e.search_space_id
{"ss_id": ss_id}, WHERE i.role_id = a.id
).fetchone() AND a.name = 'Admin'
AND e.name = 'Editor'
""")
)
editor_role = connection.execute( # Step 3: Delete all Admin roles (BULK)
sa.text(""" connection.execute(
SELECT id FROM search_space_roles sa.text("""
WHERE search_space_id = :ss_id AND name = 'Editor' DELETE FROM search_space_roles
"""), WHERE name = 'Admin' AND is_system_role = TRUE
{"ss_id": ss_id}, """)
).fetchone() )
if admin_role and editor_role: # Step 4: Update Editor permissions for all search spaces (BULK)
admin_role_id = admin_role[0]
editor_role_id = editor_role[0]
# Step 2: Move all memberships from Admin to Editor
connection.execute(
sa.text("""
UPDATE search_space_memberships
SET role_id = :editor_role_id
WHERE role_id = :admin_role_id
"""),
{"editor_role_id": editor_role_id, "admin_role_id": admin_role_id},
)
# Step 3: Move all invites from Admin to Editor
connection.execute(
sa.text("""
UPDATE search_space_invites
SET role_id = :editor_role_id
WHERE role_id = :admin_role_id
"""),
{"editor_role_id": editor_role_id, "admin_role_id": admin_role_id},
)
# Step 4: Delete the Admin role
connection.execute(
sa.text("""
DELETE FROM search_space_roles
WHERE id = :admin_role_id
"""),
{"admin_role_id": admin_role_id},
)
# Step 5: Update Editor permissions for all search spaces
editor_perms_literal = ( editor_perms_literal = (
"ARRAY[" + ",".join(f"'{p}'" for p in NEW_EDITOR_PERMISSIONS) + "]::TEXT[]" "ARRAY[" + ",".join(f"'{p}'" for p in NEW_EDITOR_PERMISSIONS) + "]::TEXT[]"
) )
@ -136,7 +115,7 @@ def upgrade():
""") """)
) )
# Step 6: Update Viewer permissions for all search spaces # Step 5: Update Viewer permissions for all search spaces (BULK)
viewer_perms_literal = ( viewer_perms_literal = (
"ARRAY[" + ",".join(f"'{p}'" for p in NEW_VIEWER_PERMISSIONS) + "]::TEXT[]" "ARRAY[" + ",".join(f"'{p}'" for p in NEW_VIEWER_PERMISSIONS) + "]::TEXT[]"
) )