"""Core API base views — tenant-aware CBV building blocks. All concrete API views in the platform should inherit from one of these base classes instead of DRF's ``APIView`` directly. They handle the repetitive boilerplate: * Asserting that ``request.tenant`` is present (injected by :class:`~simorgh.core.middleware.TenantMiddleware`). * Returning a uniform 404 / 403 response. * Wiring in :class:`~simorgh.core.api.pagination.StandardPagination` by default. Usage:: from simorgh.core.api.views import TenantListCreateView class DocumentListCreate(TenantListCreateView): required_permission = "dms.document.list" def get_queryset(self): return Document.objects.filter(tenant=self.tenant) def create_object(self, data): return services.create_document(tenant=self.tenant, **data) """ from __future__ import annotations from typing import Any from rest_framework import status from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.views import APIView from simorgh.core.api.pagination import StandardPagination __all__ = [ "TenantAPIView", "TenantListCreateView", "TenantRetrieveUpdateDestroyView", ] class TenantAPIView(APIView): """Base class for all tenant-scoped API views. Subclasses gain: * ``self.tenant`` — the current :class:`~simorgh.apps.tenants.models.Tenant` resolved by :class:`~simorgh.core.middleware.TenantMiddleware`. * ``self.tenant_404`` — a pre-built 404 response for missing tenants. * DRF ``IsAuthenticated`` permission enforced by default. Override :meth:`get_permissions` to customise the permission classes. """ permission_classes = [IsAuthenticated] # Set on the concrete view to trigger HasPermission checking. required_permission: str | None = None def get_permissions(self): permissions = super().get_permissions() if self.required_permission: from simorgh.apps.iam.permissions import HasPermission permissions.append(HasPermission()) return permissions def initialize_request(self, request: Request, *args: Any, **kwargs: Any) -> Request: request = super().initialize_request(request, *args, **kwargs) return request def initial(self, request: Request, *args: Any, **kwargs: Any) -> None: super().initial(request, *args, **kwargs) tenant = getattr(request, "tenant", None) if tenant is None: # Attach sentinel so dispatch() can detect it early. request._tenant_missing = True # type: ignore[attr-defined] else: self.tenant = tenant def dispatch(self, request: Request, *args: Any, **kwargs: Any) -> Response: response = super().dispatch(request, *args, **kwargs) return response def _no_tenant(self) -> Response: return Response( {"error": {"code": "tenant_not_found", "message": "Tenant not found."}}, status=status.HTTP_404_NOT_FOUND, ) def _forbidden(self, message: str = "Forbidden.") -> Response: return Response( {"error": {"code": "permission_denied", "message": message}}, status=status.HTTP_403_FORBIDDEN, ) def _not_found(self, resource: str = "Object") -> Response: return Response( {"error": {"code": "not_found", "message": f"{resource} not found."}}, status=status.HTTP_404_NOT_FOUND, ) def get_tenant(self) -> Any | None: """Return ``request.tenant`` or ``None``.""" return getattr(self.request, "tenant", None) class TenantListCreateView(TenantAPIView): """Base view for ``GET`` (list) + ``POST`` (create) on a tenant resource. Subclasses **must** implement: * :meth:`get_queryset` — return the base ``QuerySet`` for this tenant. * :meth:`serialize_object` — convert a model instance to a dict. Subclasses **may** implement: * :meth:`create_object` — handle POST body and return new instance. * :meth:`get_filters` — return extra ``filter()`` kwargs from query params. Pagination is applied automatically using :class:`StandardPagination`. """ pagination_class = StandardPagination # ----------------------------------------------------------------------- # Hooks — override in concrete views # ----------------------------------------------------------------------- def get_queryset(self): """Return the base (unfiltered) queryset for the current tenant.""" raise NotImplementedError(f"{self.__class__.__name__}.get_queryset() must be defined.") def serialize_object(self, obj: Any) -> dict: """Convert a model instance to a JSON-serializable dict.""" raise NotImplementedError(f"{self.__class__.__name__}.serialize_object() must be defined.") def create_object(self, data: dict) -> Any: """Create and return a new instance from validated *data*. Override to call the appropriate service function. The default implementation raises ``NotImplementedError`` so POST returns 405 unless overridden. """ raise NotImplementedError(f"{self.__class__.__name__}.create_object() must be defined.") def get_filters(self) -> dict: """Return extra ``QuerySet.filter()`` kwargs from request params. Example:: def get_filters(self): status = self.request.query_params.get("status") return {"status": status} if status else {} """ return {} # ----------------------------------------------------------------------- # HTTP handlers # ----------------------------------------------------------------------- def get(self, request: Request, *args: Any, **kwargs: Any) -> Response: tenant = self.get_tenant() if tenant is None: return self._no_tenant() qs = self.get_queryset() filters = self.get_filters() if filters: qs = qs.filter(**filters) paginator = self.pagination_class() page = paginator.paginate_queryset(qs, request, view=self) if page is not None: data = [self.serialize_object(obj) for obj in page] return paginator.get_paginated_response(data) return Response({"results": [self.serialize_object(obj) for obj in qs]}) def post(self, request: Request, *args: Any, **kwargs: Any) -> Response: tenant = self.get_tenant() if tenant is None: return self._no_tenant() # Allow if: staff user, required_permission set, or PermissionCheckMixin handles it has_mixin_guard = getattr(self, "_uses_permission_check_mixin", False) if not request.user.is_staff and not self.required_permission and not has_mixin_guard: return self._forbidden() try: obj = self.create_object(request.data) except NotImplementedError: return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) except (ValueError, TypeError) as exc: return Response( {"error": {"code": "invalid_input", "message": str(exc)}}, status=status.HTTP_400_BAD_REQUEST, ) return Response(self.serialize_object(obj), status=status.HTTP_201_CREATED) class TenantRetrieveUpdateDestroyView(TenantAPIView): """Base view for ``GET`` / ``PATCH`` / ``DELETE`` on a single tenant resource. Subclasses **must** implement: * :meth:`get_object` — return a single instance (raise ``DoesNotExist`` if missing). * :meth:`serialize_object` — convert a model instance to a dict. Subclasses **may** implement: * :meth:`update_object` — handle PATCH and return updated instance. * :meth:`destroy_object` — handle DELETE (default calls ``obj.delete()``). """ # ----------------------------------------------------------------------- # Hooks # ----------------------------------------------------------------------- def get_object(self, pk: int) -> Any: raise NotImplementedError(f"{self.__class__.__name__}.get_object() must be defined.") def serialize_object(self, obj: Any) -> dict: raise NotImplementedError(f"{self.__class__.__name__}.serialize_object() must be defined.") def update_object(self, obj: Any, data: dict) -> Any: raise NotImplementedError(f"{self.__class__.__name__}.update_object() must be defined.") def destroy_object(self, obj: Any) -> None: obj.delete() # ----------------------------------------------------------------------- # HTTP handlers # ----------------------------------------------------------------------- def _resolve(self, pk: int): """Return ``(obj, None)`` or ``(None, error_response)``.""" tenant = self.get_tenant() if tenant is None: return None, self._no_tenant() try: return self.get_object(pk), None except Exception as exc: if exc.__class__.__name__ == "DoesNotExist": return None, self._not_found() raise def get(self, request: Request, pk: int, *args: Any, **kwargs: Any) -> Response: obj, err = self._resolve(pk) if err: return err return Response(self.serialize_object(obj)) def patch(self, request: Request, pk: int, *args: Any, **kwargs: Any) -> Response: obj, err = self._resolve(pk) if err: return err if not request.user.is_staff and not self.required_permission: return self._forbidden() try: obj = self.update_object(obj, request.data) except NotImplementedError: return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) except (ValueError, TypeError) as exc: return Response( {"error": {"code": "invalid_input", "message": str(exc)}}, status=status.HTTP_400_BAD_REQUEST, ) return Response(self.serialize_object(obj)) def delete(self, request: Request, pk: int, *args: Any, **kwargs: Any) -> Response: obj, err = self._resolve(pk) if err: return err if not request.user.is_staff and not self.required_permission: return self._forbidden() self.destroy_object(obj) return Response(status=status.HTTP_204_NO_CONTENT)