Establish access rights via migration

src/api/__init__.py

@@ -13,9 +13,7 @@
 from fastapi.routing import APIRoute
 from pydantic import BaseModel
 
-from ..database import engine
 from ..deployment.monitors.health import vm_monitor
-from ._util.role import create_access_rights_if_emtpy
 from .backup import router as backup_router
 from .backupmonitor import run_backup_monitor
 from .organization import api as organization_api
@@ -194,9 +192,6 @@ async def _populate_db():
     config.set_main_option("script_location", migrations_path)
     command.upgrade(config, "head")
 
-    async with engine.begin() as conn:
-        await create_access_rights_if_emtpy(conn)
-
 
 def _use_route_names_as_operation_ids(app: FastAPI) -> None:
     """

src/api/_util/role.py

@@ -1,13 +1,9 @@
-from typing import get_args
-
-from sqlalchemy.ext.asyncio import AsyncConnection
-from sqlmodel import insert, select
-from ulid import ULID
+from sqlmodel import select
 
 from ..._util import Identifier
 from ...database import SessionDep
 from ...models.branch import Branch
-from ...models.role import AccessRight, AccessRightPublic, Organization, Role, RoleAccessRight, RoleType, RoleUserLink
+from ...models.role import AccessRight, Organization, Role, RoleAccessRight, RoleType, RoleUserLink
 
 
 async def clone_user_role_assignment(
@@ -34,33 +30,6 @@ async def clone_user_role_assignment(
     await session.refresh(target)
 
 
-def get_role_type(access_right: AccessRightPublic) -> RoleType:
-    name = str(access_right)
-    if name.startswith("org:"):
-        return RoleType.organization
-    elif name.startswith("env:"):
-        return RoleType.environment
-    elif name.startswith("project:"):
-        return RoleType.project
-    elif name.startswith("branch:"):
-        return RoleType.branch
-    else:
-        raise ValueError(f"Invalid access right: {name}")
-
-
-async def create_access_rights_if_emtpy(conn: AsyncConnection):
-    result = await conn.execute(select(AccessRight))
-    if len(list(result.scalars().all())) == 0:
-        for access_right_public in get_args(AccessRightPublic):
-            await conn.execute(
-                insert(AccessRight).values(
-                    id=ULID(),
-                    entry=access_right_public,
-                    role_type=get_role_type(access_right_public).name,
-                )
-            )
-
-
 async def get_access_rights(session: SessionDep) -> list[AccessRight]:
     result = await session.execute(select(AccessRight))
     return list(result.scalars().all())

src/models/migrations/versions/f76768ab95d8_establish_access_rights.py

@@ -0,0 +1,111 @@
+"""establish-access-rights
+
+Revision ID: f76768ab95d8
+Revises: 6726a34eb0ec
+Create Date: 2026-03-26 08:20:53.641921
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'f76768ab95d8'
+down_revision: Union[str, Sequence[str], None] = '6726a34eb0ec'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+_ACCESS_RIGHTS = [
+    ("org:owner:admin", "organization"),
+    ("org:settings:read", "organization"),
+    ("org:settings:admin", "organization"),
+    ("org:auth:read", "organization"),
+    ("org:auth:admin", "organization"),
+    ("org:backup:read", "organization"),
+    ("org:backup:update", "organization"),
+    ("org:backup:create", "organization"),
+    ("org:backup:delete", "organization"),
+    ("org:metering:read", "organization"),
+    ("org:role:read", "organization"),
+    ("org:role:admin", "organization"),
+    ("org:user:read", "organization"),
+    ("org:user:admin", "organization"),
+    ("org:role-assign:read", "organization"),
+    ("org:role-assign:admin", "organization"),
+    ("org:projects:read", "organization"),
+    ("org:projects:write", "organization"),
+    ("org:projects:create", "organization"),
+    ("org:projects:stop", "organization"),
+    ("org:projects:pause", "organization"),
+    ("org:projects:delete", "organization"),
+    ("org:projects:apikeys", "organization"),
+    ("org:limits:read", "organization"),
+    ("org:limits:admin", "organization"),
+    ("env:db:admin", "environment"),
+    ("env:projects:read", "environment"),
+    ("env:projects:admin", "environment"),
+    ("env:backup:read", "environment"),
+    ("env:backup:admin", "environment"),
+    ("env:projects:write", "environment"),
+    ("env:projects:create", "environment"),
+    ("env:role-assign:read", "environment"),
+    ("env:role-assign:admin", "environment"),
+    ("env:projects:stop", "environment"),
+    ("env:projects:pause", "environment"),
+    ("env:projects:delete", "environment"),
+    ("env:projects:getkeys", "environment"),
+    ("project:settings:read", "project"),
+    ("project:settings:write", "project"),
+    ("project:role-assign:read", "project"),
+    ("project:role-assign:admin", "project"),
+    ("project:branches:create", "project"),
+    ("project:branches:delete", "project"),
+    ("project:branches:stop", "project"),
+    ("branch:settings:read", "branch"),
+    ("branch:settings:admin", "branch"),
+    ("branch:role-assign:read", "branch"),
+    ("branch:role-assign:admin", "branch"),
+    ("branch:auth:read", "branch"),
+    ("branch:auth:admin", "branch"),
+    ("branch:api:getkeys", "branch"),
+    ("branch:replicate:read", "branch"),
+    ("branch:replicate:admin", "branch"),
+    ("branch:import:read", "branch"),
+    ("branch:import:admin", "branch"),
+    ("branch:logging:read", "branch"),
+    ("branch:monitoring:read", "branch"),
+    ("branch:db:admin", "branch"),
+    ("branch:rls:read", "branch"),
+    ("branch:rls:admin", "branch"),
+    ("branch:edge:read", "branch"),
+    ("branch:edge:admin", "branch"),
+    ("branch:rt:read", "branch"),
+    ("branch:rt:admin", "branch"),
+]
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    conn = op.get_bind()
+    for entry, role_type in _ACCESS_RIGHTS:
+        conn.execute(
+            sa.text(
+                "INSERT INTO accessright (id, entry, role_type) "
+                "SELECT gen_random_uuid(), :entry, :role_type "
+                "WHERE NOT EXISTS (SELECT 1 FROM accessright WHERE entry = :entry_check)"
+            ),
+            {"entry": entry, "role_type": role_type, "entry_check": entry},
+        )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    conn = op.get_bind()
+    conn.execute(
+        sa.text("DELETE FROM accessright WHERE entry = ANY(:entries)").bindparams(
+            sa.bindparam("entries", value=[e for e, _ in _ACCESS_RIGHTS], type_=sa.ARRAY(sa.String))
+        )
+    )

src/models/role.py

@@ -1,8 +1,8 @@
 from enum import Enum as PyEnum
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Annotated, Literal
 from uuid import UUID
 
-from pydantic import BaseModel
+from pydantic import BaseModel, StringConstraints
 from sqlalchemy.ext.asyncio import AsyncAttrs
 from sqlmodel import Field, Relationship
 
@@ -17,78 +17,7 @@
 RoleTypePublic = Literal["organization", "environment", "project", "branch"]
 
 
-AccessRightPublic = Literal[
-    "org:owner:admin",
-    "org:settings:read",
-    "org:settings:admin",
-    "org:auth:read",
-    "org:auth:admin",
-    "org:backup:read",
-    "org:backup:update",
-    "org:backup:create",
-    "org:backup:delete",
-    "org:metering:read",
-    "org:role:read",
-    "org:role:admin",
-    "org:user:read",
-    "org:user:admin",
-    "org:role-assign:read",
-    "org:role-assign:admin",
-    "org:projects:read",
-    "org:projects:write",
-    "org:projects:create",
-    "org:projects:stop",
-    "org:projects:pause",
-    "org:projects:delete",
-    "org:projects:pause",
-    "org:projects:apikeys",
-    "org:limits:read",
-    "env:db:admin",
-    "env:projects:read",
-    "env:projects:admin",
-    "org:limits:admin",
-    "env:backup:read",
-    "env:backup:admin",
-    "env:projects:read",
-    "env:projects:write",
-    "env:projects:create",
-    "env:role-assign:read",
-    "env:role-assign:admin",
-    "env:projects:stop",
-    "env:projects:pause",
-    "env:projects:delete",
-    "env:projects:getkeys",
-    "env:db:admin",
-    "env:projects:read",
-    "env:projects:admin",
-    "project:settings:read",
-    "project:settings:write",
-    "project:role-assign:read",
-    "project:role-assign:admin",
-    "project:branches:create",
-    "project:branches:delete",
-    "project:branches:stop",
-    "branch:settings:read",
-    "branch:settings:admin",
-    "branch:role-assign:read",
-    "branch:role-assign:admin",
-    "branch:auth:read",
-    "branch:auth:admin",
-    "branch:api:getkeys",
-    "branch:replicate:read",
-    "branch:replicate:admin",
-    "branch:import:read",
-    "branch:import:admin",
-    "branch:logging:read",
-    "branch:monitoring:read",
-    "branch:db:admin",
-    "branch:rls:read",
-    "branch:rls:admin",
-    "branch:edge:read",
-    "branch:edge:admin",
-    "branch:rt:read",
-    "branch:rt:admin",
-]
+AccessRightPublic = Annotated[str, StringConstraints(pattern=r"^(org|env|project|branch):[a-z][a-z\-]*:[a-z][a-z\-]*$")]
 
 
 class RoleType(PyEnum):