refactor(b3): dynamic entity registry, custom_fields permissions decoupled from contacts, write perms generated from registry

app/core/permission_registry.py

@@ -54,6 +54,8 @@ CORE_PERMISSIONS: list[dict[str, str]] = [
     {"key": "taxes:write", "label": "Taxes: Write", "category": "core", "module": "taxes"},
     {"key": "currencies:read", "label": "Currencies: Read", "category": "core", "module": "currencies"},
     {"key": "currencies:write", "label": "Currencies: Write", "category": "core", "module": "currencies"},
+    {"key": "custom_fields:read", "label": "Custom Fields: Read", "category": "core", "module": "custom_fields"},
+    {"key": "custom_fields:write", "label": "Custom Fields: Write", "category": "core", "module": "custom_fields"},
     {"key": "import_export:read", "label": "Import/Export: Read", "category": "core", "module": "import_export"},
     {"key": "import_export:write", "label": "Import/Export: Write", "category": "core", "module": "import_export"},
     {"key": "workspaces:read", "label": "Workspaces: Read", "category": "core", "module": "workspaces"},

app/deps.py

@@ -17,8 +17,9 @@ from app.core.db import get_db, set_tenant_context, set_user_context
 
 logger = logging.getLogger(__name__)
 
-# Known write-permission modules — used by require_write() to check
+# Legacy fallback list — used by require_write() only when the permission
-# specific permissions instead of broad wildcards like *:write
+# registry is not initialized. The live source of truth is generated from
+# the registry (see _get_write_permissions, ARCH-022).
 _WRITE_PERMISSIONS = [
     "users:write",
     "roles:write",
@@ -35,6 +36,30 @@ _WRITE_PERMISSIONS = [
 ]
 
 
+def _get_write_permissions() -> list[str]:
+    """Return all known ``module:write`` permission keys (ARCH-022).
+
+    Generated from the permission registry so plugin write permissions are
+    picked up automatically without touching this file. Falls back to the
+    static legacy list when the registry is unavailable/uninitialized.
+    """
+    try:
+        from app.core.permission_registry import get_permission_registry
+
+        registry = get_permission_registry()
+        if getattr(registry, "_initialized", False):
+            perms = [
+                entry["key"]
+                for entry in registry.get_all()
+                if entry["key"].endswith(":write")
+            ]
+            if perms:
+                return sorted(perms)
+    except Exception:
+        pass
+    return list(_WRITE_PERMISSIONS)
+
+
 async def get_redis_dep() -> aioredis.Redis:
     """FastAPI dependency for Redis client."""
     return get_redis()
@@ -261,7 +286,7 @@ async def require_write(
     # Check via permission system for specific write permissions
     from app.core.permissions import check_permission
 
-    for perm in _WRITE_PERMISSIONS:
+    for perm in _get_write_permissions():
         if check_permission(current_user, perm):
             return current_user
 

app/routes/custom_field_definitions.py

@@ -22,7 +22,7 @@ router = APIRouter(prefix="/api/v1/custom-fields", tags=["custom-fields-definiti
 @router.get(
     "/definitions",
     response_model=list[CustomFieldDefinitionResponse],
-    dependencies=[Depends(require_permission("contacts:read"))],
+    dependencies=[Depends(require_permission("custom_fields:read"))],
 )
 async def list_definitions(
     entity: str | None = Query(None, description="Filter by entity type (e.g. 'contact', 'company')"),
@@ -39,7 +39,7 @@ async def list_definitions(
     "/definitions",
     response_model=CustomFieldDefinitionResponse,
     status_code=201,
-    dependencies=[Depends(require_permission("contacts:write"))],
+    dependencies=[Depends(require_permission("custom_fields:write"))],
 )
 async def create_definition(
     body: CustomFieldDefinitionCreate,
@@ -58,7 +58,7 @@ async def create_definition(
 @router.patch(
     "/definitions/{definition_id}",
     response_model=CustomFieldDefinitionResponse,
-    dependencies=[Depends(require_permission("contacts:write"))],
+    dependencies=[Depends(require_permission("custom_fields:write"))],
 )
 async def update_definition(
     definition_id: str,
@@ -90,7 +90,7 @@ async def update_definition(
 @router.delete(
     "/definitions/{definition_id}",
     status_code=204,
-    dependencies=[Depends(require_permission("contacts:write"))],
+    dependencies=[Depends(require_permission("custom_fields:write"))],
 )
 async def delete_definition(
     definition_id: str,

app/routes/entity_permissions.py

@@ -235,22 +235,45 @@ async def list_all_permissions(
 async def list_entity_registry(
     current_user: dict = Depends(get_current_user),
 ):
-    """List all registered entity types that support permissions."""
+    """List all registered entity types that support permissions.
-    # Static list for now — will be dynamic from Permission Registry in Sprint 4
+
-    entity_types = [
+    Generated dynamically from the entity model registry (ARCH-016): core
-        {"entity_type": "contact", "label": "Kontakte", "table": "contacts"},
+    models plus every entity registered by an active plugin at activation
-        {"entity_type": "contact_folder", "label": "Ordner", "table": "contact_folders"},
+    time. Plugin entities appear automatically without touching this file.
-        {"entity_type": "address", "label": "Adressen", "table": "addresses"},
+    """
-        {"entity_type": "attachment", "label": "Anhänge", "table": "attachments"},
+    from app.services.entity_permission_service import ENTITY_MODELS
-        {"entity_type": "bank_account", "label": "Bankkonten", "table": "bank_accounts"},
+
-        {"entity_type": "workflow", "label": "Workflows", "table": "workflows"},
+    # Human-readable labels for known types; plugin entities fall back to a
-        {"entity_type": "sequence", "label": "Sequenzen", "table": "sequences"},
+    # title-cased entity_type so they are still presentable.
-        {"entity_type": "saved_filter", "label": "Gespeicherte Filter", "table": "saved_filters"},
+    _labels = {
-        {"entity_type": "saved_view", "label": "Gespeicherte Ansichten", "table": "saved_views"},
+        "contact": "Kontakte",
-        {"entity_type": "webhook", "label": "Webhooks", "table": "webhooks"},
+        "contact_folder": "Ordner",
-        {"entity_type": "notification", "label": "Benachrichtigungen", "table": "notifications"},
+        "address": "Adressen",
-        {"entity_type": "custom_field_definition", "label": "Custom Fields", "table": "custom_field_definitions"},
+        "attachment": "Anhänge",
-    ]
+        "bank_account": "Bankkonten",
+        "workflow": "Workflows",
+        "sequence": "Sequenzen",
+        "saved_filter": "Gespeicherte Filter",
+        "saved_view": "Gespeicherte Ansichten",
+        "webhook": "Webhooks",
+        "notification": "Benachrichtigungen",
+        "custom_field_definition": "Custom Fields",
+    }
+
+    seen_models: set[int] = set()
+    entity_types: list[dict[str, str]] = []
+    for entity_type, model_class in ENTITY_MODELS.items():
+        # Skip aliases pointing at the same model (contact/contacts/company)
+        if id(model_class) in seen_models:
+            continue
+        seen_models.add(id(model_class))
+        table = getattr(model_class, "__tablename__", f"{entity_type}s")
+        entity_types.append({
+            "entity_type": entity_type,
+            "label": _labels.get(entity_type, entity_type.replace("_", " ").title()),
+            "table": table,
+        })
+    entity_types.sort(key=lambda e: e["entity_type"])
     return {"items": entity_types, "total": len(entity_types)}
 
 

app/services/entity_permission_service.py

@@ -34,6 +34,7 @@ from app.models.contact_folder import ContactFolder
 from app.models.custom_field_definition import CustomFieldDefinition
 from app.models.entity_permission import EntityPermission
 from app.models.group import Group, UserGroup
+from app.models.notification import Notification
 from app.models.role import Role
 from app.models.saved_filter import SavedFilter
 from app.models.saved_view import SavedView
@@ -65,6 +66,7 @@ ENTITY_MODELS: dict[str, type] = {
     "saved_filter": SavedFilter,
     "saved_view": SavedView,
     "webhook": Webhook,
+    "notification": Notification,
     "custom_field_definition": CustomFieldDefinition,
     "contact_folder": ContactFolder,
 }