Make resource_registry work without rbac

Author: john-westcott-iv

ansible_base/resource_registry/registry.py

@@ -1,6 +1,7 @@
 from collections import namedtuple
 from typing import List, Optional
 
+from django.conf import settings
 from django.contrib.auth import authenticate
 from django.utils.translation import gettext_lazy as _
 
@@ -23,12 +24,16 @@ class ServiceAPIConfig:
     This will be the interface for configuring the resource registry for each service.
     """
 
-    _default_resource_processors = {
-        "shared.team": ResourceTypeProcessor,
-        "shared.organization": ResourceTypeProcessor,
-        "shared.user": ResourceTypeProcessor,
-        "shared.roledefinition": RoleDefinitionProcessor,
-    }
+    @classmethod
+    def _get_default_resource_processors(cls):
+        processors = {
+            "shared.team": ResourceTypeProcessor,
+            "shared.organization": ResourceTypeProcessor,
+            "shared.user": ResourceTypeProcessor,
+        }
+        if 'ansible_base.rbac' in settings.INSTALLED_APPS:
+            processors["shared.roledefinition"] = RoleDefinitionProcessor
+        return processors
 
     custom_resource_processors = {}
 
@@ -43,7 +48,7 @@ def authenticate_local_user(username: str, password: str):
 
     @classmethod
     def get_processor(cls, resource_type):
-        combined_processors = {**cls._default_resource_processors, **cls.custom_resource_processors}
+        combined_processors = {**cls._get_default_resource_processors(), **cls.custom_resource_processors}
         return combined_processors[resource_type]
 
 

ansible_base/resource_registry/rest_client.py

@@ -6,9 +6,17 @@
 import requests
 import urllib3
 from django.apps import apps
+from django.conf import settings
 
 from ansible_base.resource_registry.resource_server import get_resource_server_config, get_service_token
 
+
+def _check_rbac_installed():
+    """Check if ansible_base.rbac is installed and raise RuntimeError if not."""
+    if 'ansible_base.rbac' not in settings.INSTALLED_APPS:
+        raise RuntimeError("This operation requires ansible_base.rbac to be installed")
+
+
 ResourceRequestBody = namedtuple(
     "ResourceRequestBody",
     ["ansible_id", "service_id", "is_partially_migrated", "resource_type", "resource_data"],
@@ -166,26 +174,31 @@ def get_resource_type_manifest(self, name, filters: Optional[dict] = None):
 
     # RBAC related methods
     def list_role_types(self, filters: Optional[dict] = None):
+        _check_rbac_installed()
         return self._make_request("get", "role-types/", params=filters)
 
     def list_role_permissions(self, filters: Optional[dict] = None):
+        _check_rbac_installed()
         return self._make_request("get", "role-permissions/", params=filters)
 
     def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: Optional[dict] = None):
         """List user role assignments."""
+        _check_rbac_installed()
         params = (filters or {}).copy()
         if user_ansible_id is not None:
             params['user_ansible_id'] = user_ansible_id
         return self._make_request("get", "role-user-assignments/", params=params)
 
     def list_team_assignments(self, team_ansible_id: Optional[str] = None, filters: Optional[dict] = None):
         """List team role assignments."""
+        _check_rbac_installed()
         params = (filters or {}).copy()
         if team_ansible_id is not None:
             params['team_ansible_id'] = team_ansible_id
         return self._make_request("get", "role-team-assignments/", params=params)
 
     def sync_assignment(self, assignment):
+        _check_rbac_installed()
         from ansible_base.rbac.service_api.serializers import ServiceRoleTeamAssignmentSerializer, ServiceRoleUserAssignmentSerializer
 
         if assignment._meta.model_name == 'roleuserassignment':
@@ -196,6 +209,7 @@ def sync_assignment(self, assignment):
         return self._sync_assignment(serializer.data)
 
     def sync_unassignment(self, role_definition, actor, content_object):
+        _check_rbac_installed()
         data = {'role_definition': role_definition.name}
         data[f'{actor._meta.model_name}_ansible_id'] = str(actor.resource.ansible_id)
 
@@ -214,6 +228,7 @@ def sync_unassignment(self, role_definition, actor, content_object):
 
     def sync_object_deletion(self, content_object):
         """Sync object deletion to Gateway for cleanup of all related role assignments"""
+        _check_rbac_installed()
         from ansible_base.rbac.models import DABContentType
 
         # Get the content type information

ansible_base/resource_registry/shared_types.py

@@ -1,7 +1,7 @@
+from django.conf import settings
 from rest_framework import serializers
 from rest_framework.exceptions import ValidationError
 
-from ansible_base.rbac.models import DABContentType, DABPermission
 from ansible_base.resource_registry.utils.resource_type_serializers import AnsibleResourceForeignKeyField, SharedResourceTypeSerializer
 from ansible_base.resource_registry.utils.sso_provider import get_sso_provider_server
 
@@ -84,6 +84,10 @@ class LenientPermissionSlugListField(serializers.ListField):
     child = serializers.CharField()
 
     def to_internal_value(self, data):
+        if 'ansible_base.rbac' not in settings.INSTALLED_APPS:
+            raise RuntimeError("LenientPermissionSlugListField requires ansible_base.rbac to be installed")
+        from ansible_base.rbac.models import DABPermission
+
         data = super().to_internal_value(data)
         return list(DABPermission.objects.filter(api_slug__in=data))
 
@@ -98,14 +102,24 @@ class RoleDefinitionType(SharedResourceTypeSerializer):
     name = serializers.CharField()
     description = serializers.CharField(default="", allow_blank=True)
     managed = serializers.BooleanField()
-    content_type = serializers.SlugRelatedField(
-        slug_field='api_slug',
-        queryset=DABContentType.objects.all(),
-        allow_null=True,
-        default=None,
-    )
     permissions = LenientPermissionSlugListField()
 
+    def __init__(self, *args, **kwargs):
+        if 'ansible_base.rbac' not in settings.INSTALLED_APPS:
+            raise RuntimeError("RoleDefinitionType requires ansible_base.rbac to be installed")
+
+        super().__init__(*args, **kwargs)
+
+        # Set up content_type field only when rbac is available
+        from ansible_base.rbac.models import DABContentType
+
+        self.fields['content_type'] = serializers.SlugRelatedField(
+            slug_field='api_slug',
+            queryset=DABContentType.objects.all(),
+            allow_null=True,
+            default=None,
+        )
+
     def is_valid(self, raise_exception=False):
         try:
             return super().is_valid(raise_exception=raise_exception)

ansible_base/resource_registry/tasks/sync.py

@@ -18,14 +18,15 @@
 from django.db.utils import Error, IntegrityError
 from requests import HTTPError
 
-from ansible_base.rbac.models.role import AssignmentBase, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ansible_base.resource_registry.models import Resource, ResourceType
 from ansible_base.resource_registry.models.service_identifier import service_id
 from ansible_base.resource_registry.registry import get_registry
 from ansible_base.resource_registry.rest_client import ResourceAPIClient, get_resource_server_client
 
 logger = logging.getLogger('ansible_base.resources_api.tasks.sync')
 
+_is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS
+
 
 class ManifestNotFound(HTTPError):
     """Raise when server returns 404 for a manifest"""
@@ -139,7 +140,9 @@ def fetch_manifest(
     return [ManifestItem(**row) for row in csv_reader]
 
 
-def get_ansible_id_or_pk(assignment: AssignmentBase) -> str:
+def get_ansible_id_or_pk(assignment) -> str:
+    if not _is_rbac_installed:
+        raise RuntimeError("get_ansible_id_or_pk requires ansible_base.rbac to be installed")
     # For object-scoped assignments, try to get the object's ansible_id
     if assignment.content_type.model in ('organization', 'team'):
         object_resource = Resource.objects.filter(object_id=assignment.object_id, content_type__model=assignment.content_type.model).first()
@@ -153,7 +156,9 @@ def get_ansible_id_or_pk(assignment: AssignmentBase) -> str:
     return str(ansible_id_or_pk)
 
 
-def get_content_object(role_definition: RoleDefinition, assignment_tuple: AssignmentTuple) -> Any:
+def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> Any:
+    if not _is_rbac_installed:
+        raise RuntimeError("get_content_object requires ansible_base.rbac to be installed")
     content_object = None
     if role_definition.content_type.model in ('organization', 'team'):
         object_resource = Resource.objects.get(ansible_id=assignment_tuple.ansible_id_or_pk)
@@ -167,6 +172,8 @@ def get_content_object(role_definition: RoleDefinition, assignment_tuple: Assign
 
 def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple]:
     """Fetch remote assignments from the resource server and convert to tuples."""
+    if not _is_rbac_installed:
+        raise RuntimeError("get_remote_assignments requires ansible_base.rbac to be installed")
     assignments = set()
 
     # Fetch user assignments with pagination
@@ -238,6 +245,10 @@ def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple
 
 def get_local_assignments() -> set[AssignmentTuple]:
     """Get local assignments and convert to tuples."""
+    if not _is_rbac_installed:
+        raise RuntimeError("get_local_assignments requires ansible_base.rbac to be installed")
+    from ansible_base.rbac.models.role import RoleTeamAssignment, RoleUserAssignment
+
     assignments = set()
 
     # Get user assignments
@@ -294,6 +305,10 @@ def get_local_assignments() -> set[AssignmentTuple]:
 
 def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool:
     """Delete a local assignment based on the tuple."""
+    if not _is_rbac_installed:
+        raise RuntimeError("delete_local_assignment requires ansible_base.rbac to be installed")
+    from ansible_base.rbac.models.role import RoleDefinition
+
     try:
         role_definition = RoleDefinition.objects.get(name=assignment_tuple.role_definition_name)
 
@@ -320,6 +335,10 @@ def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool:
 
 def create_local_assignment(assignment_tuple: AssignmentTuple) -> bool:
     """Create a local assignment based on the tuple."""
+    if not _is_rbac_installed:
+        raise RuntimeError("create_local_assignment requires ansible_base.rbac to be installed")
+    from ansible_base.rbac.models.role import RoleDefinition
+
     try:
         role_definition = RoleDefinition.objects.get(name=assignment_tuple.role_definition_name)
 
@@ -694,6 +713,10 @@ def _sync_assignments(self):
         if not self.sync_assignments:
             return
 
+        if not _is_rbac_installed:
+            self.write(">>> Skipping role assignments sync (rbac not installed)")
+            return
+
         self.write(">>> Syncing role assignments")
 
         try:

test_app/resource_api.py

@@ -1,9 +1,9 @@
+from django.conf import settings
 from django.contrib.auth import get_user_model
 
 from ansible_base.authentication.models import Authenticator
-from ansible_base.rbac.models import RoleDefinition
 from ansible_base.resource_registry.registry import ResourceConfig, ServiceAPIConfig, SharedResource
-from ansible_base.resource_registry.shared_types import OrganizationType, RoleDefinitionType, TeamType, UserType
+from ansible_base.resource_registry.shared_types import OrganizationType, TeamType, UserType
 from ansible_base.resource_registry.utils.resource_type_processor import ResourceTypeProcessor
 from test_app.models import Organization, Original1, Proxy2, ResourceMigrationTestModel, Team
 
@@ -38,13 +38,21 @@ class APIConfig(ServiceAPIConfig):
         Organization,
         shared_resource=SharedResource(serializer=OrganizationType, is_provider=False),
     ),
-    ResourceConfig(
-        RoleDefinition,
-        shared_resource=SharedResource(serializer=RoleDefinitionType, is_provider=False),
-    ),
     # Authenticators won't be a shared resource in production, but it's a convenient model to use for testing.
     ResourceConfig(Authenticator),
     ResourceConfig(ResourceMigrationTestModel),
     ResourceConfig(Original1),
     ResourceConfig(Proxy2),
 ]
+
+# Conditionally add RoleDefinition if RBAC is installed
+if 'ansible_base.rbac' in settings.INSTALLED_APPS:
+    from ansible_base.rbac.models import RoleDefinition
+    from ansible_base.resource_registry.shared_types import RoleDefinitionType
+
+    RESOURCE_LIST.append(
+        ResourceConfig(
+            RoleDefinition,
+            shared_resource=SharedResource(serializer=RoleDefinitionType, is_provider=False),
+        )
+    )