added user and facility flag

care/emr/api/viewsets/facility_flag.py

@@ -0,0 +1,101 @@
+from django.db import transaction
+from django_filters import rest_framework as filters
+from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.response import Response
+
+from care.emr.api.viewsets.base import (
+    EMRBaseViewSet,
+    EMRCreateMixin,
+    EMRDestroyMixin,
+    EMRListMixin,
+    EMRRetrieveMixin,
+    EMRUpdateMixin,
+)
+from care.emr.resources.facility_flag.spec import (
+    FacilityFlagCreateSpec,
+    FacilityFlagReadSpec,
+    FacilityFlagRetrieveSpec,
+    FacilityFlagUpdateSpec,
+)
+from care.facility.models import FacilityFlag
+from care.security.authorization.base import AuthorizationController
+from care.utils.registries.feature_flag import FlagNotFoundError, FlagRegistry, FlagType
+
+
+class FacilityFlagFilters(filters.FilterSet):
+    facility = filters.UUIDFilter(field_name="facility__external_id")
+    flag = filters.CharFilter(field_name="flag", lookup_expr="iexact")
+
+
+class FacilityFlagViewSet(
+    EMRCreateMixin,
+    EMRRetrieveMixin,
+    EMRUpdateMixin,
+    EMRListMixin,
+    EMRDestroyMixin,
+    EMRBaseViewSet,
+):
+    database_model = FacilityFlag
+    pydantic_model = FacilityFlagCreateSpec
+    pydantic_update_model = FacilityFlagUpdateSpec
+    pydantic_read_model = FacilityFlagReadSpec
+    pydantic_retrieve_model = FacilityFlagRetrieveSpec
+    filterset_class = FacilityFlagFilters
+    filter_backends = [filters.DjangoFilterBackend]
+
+    def authorize_create(self, instance):
+        if not AuthorizationController.call(
+            "can_write_facility_flag", self.request.user
+        ):
+            raise PermissionDenied(
+                "You do not have permission to create facility flags"
+            )
+
+    def authorize_update(self, request_obj, model_instance):
+        if not AuthorizationController.call(
+            "can_write_facility_flag", self.request.user
+        ):
+            raise PermissionDenied(
+                "You do not have permission to update facility flags"
+            )
+
+    def authorize_destroy(self, instance):
+        if not AuthorizationController.call(
+            "can_write_facility_flag", self.request.user
+        ):
+            raise PermissionDenied(
+                "You do not have permission to delete facility flags"
+            )
+
+    def get_queryset(self):
+        if not AuthorizationController.call(
+            "can_read_facility_flag", self.request.user
+        ):
+            raise PermissionDenied("You do not have permission to list facility flags")
+        return super().get_queryset()
+
+    def perform_create(self, instance):
+        with transaction.atomic():
+            FlagRegistry.register(FlagType.FACILITY, instance.flag)
+            super().perform_create(instance)
+
+    def perform_destroy(self, instance):
+        with transaction.atomic():
+            super().perform_destroy(instance)
+            FlagRegistry.unregister(FlagType.FACILITY, instance.flag)
+
+    @action(detail=False, methods=["GET"], url_path="available-flags")
+    def available_flags(self, request):
+        if not AuthorizationController.call(
+            "can_read_facility_flag", self.request.user
+        ):
+            raise PermissionDenied("You do not have permission to view available flags")
+
+        try:
+            flags = FlagRegistry.get_all_flags(FlagType.FACILITY)
+            return Response({"available_flags": list(flags)})
+        except FlagNotFoundError:
+            return Response(
+                {"message": "No registered flag type 'facility' found."}, status=400
+            )

care/emr/api/viewsets/user_flag.py

@@ -0,0 +1,84 @@
+from django.db import transaction
+from django_filters import rest_framework as filters
+from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.response import Response
+
+from care.emr.api.viewsets.base import (
+    EMRBaseViewSet,
+    EMRCreateMixin,
+    EMRDestroyMixin,
+    EMRListMixin,
+    EMRRetrieveMixin,
+    EMRUpdateMixin,
+)
+from care.emr.resources.user_flag.spec import (
+    UserFlagCreateSpec,
+    UserFlagReadSpec,
+    UserFlagRetrieveSpec,
+    UserFlagUpdateSpec,
+)
+from care.security.authorization.base import AuthorizationController
+from care.users.models import UserFlag
+from care.utils.registries.feature_flag import FlagNotFoundError, FlagRegistry, FlagType
+
+
+class UserFlagFilters(filters.FilterSet):
+    user = filters.UUIDFilter(field_name="user__external_id")
+    flag = filters.CharFilter(field_name="flag", lookup_expr="iexact")
+
+
+class UserFlagViewSet(
+    EMRCreateMixin,
+    EMRRetrieveMixin,
+    EMRUpdateMixin,
+    EMRListMixin,
+    EMRDestroyMixin,
+    EMRBaseViewSet,
+):
+    database_model = UserFlag
+    pydantic_model = UserFlagCreateSpec
+    pydantic_update_model = UserFlagUpdateSpec
+    pydantic_read_model = UserFlagReadSpec
+    pydantic_retrieve_model = UserFlagRetrieveSpec
+    filterset_class = UserFlagFilters
+    filter_backends = [filters.DjangoFilterBackend]
+
+    def authorize_create(self, instance):
+        if not AuthorizationController.call("can_write_user_flag", self.request.user):
+            raise PermissionDenied("You do not have permission to create user flags")
+
+    def authorize_update(self, request_obj, model_instance):
+        if not AuthorizationController.call("can_write_user_flag", self.request.user):
+            raise PermissionDenied("You do not have permission to update user flags")
+
+    def authorize_destroy(self, instance):
+        if not AuthorizationController.call("can_write_user_flag", self.request.user):
+            raise PermissionDenied("You do not have permission to delete user flags")
+
+    def get_queryset(self):
+        if not AuthorizationController.call("can_read_user_flag", self.request.user):
+            raise PermissionDenied("You do not have permission to list user flags")
+        return super().get_queryset()
+
+    def perform_create(self, instance):
+        with transaction.atomic():
+            FlagRegistry.register(FlagType.USER, instance.flag)
+            super().perform_create(instance)
+
+    def perform_destroy(self, instance):
+        with transaction.atomic():
+            super().perform_destroy(instance)
+            FlagRegistry.unregister(FlagType.USER, instance.flag)
+
+    @action(detail=False, methods=["GET"], url_path="available-flags")
+    def available_flags(self, request):
+        if not AuthorizationController.call("can_read_user_flag", self.request.user):
+            raise PermissionDenied("You do not have permission to view available flags")
+        try:
+            flags = FlagRegistry.get_all_flags(FlagType.USER)
+            return Response({"available_flags": list(flags)})
+        except FlagNotFoundError:
+            return Response(
+                {"message": "No registered flag type 'user' found."}, status=400
+            )

care/emr/resources/facility_flag/spec.py

@@ -0,0 +1,51 @@
+from pydantic import UUID4, field_validator
+
+from care.emr.resources.base import EMRResource
+from care.emr.resources.facility.spec import FacilityBareMinimumSpec
+from care.facility.models import FacilityFlag
+from care.facility.models.facility import Facility
+from care.utils.registries.feature_flag import FlagRegistry, FlagType
+from care.utils.shortcuts import get_object_or_404
+
+
+class FacilityFlagBaseSpec(EMRResource):
+    __model__ = FacilityFlag
+    __exclude__ = ["facility"]
+
+    id: UUID4 | None = None
+    flag: str
+
+
+class FacilityFlagCreateSpec(FacilityFlagBaseSpec):
+    facility: UUID4
+
+    @field_validator("flag")
+    @classmethod
+    def validate_flag_name(cls, flag_name):
+        FlagRegistry.validate_flag_name(FlagType.FACILITY, flag_name)
+        return flag_name
+
+    def perform_extra_deserialization(self, is_update, obj):
+        if not is_update:
+            obj.facility = get_object_or_404(Facility, external_id=self.facility)
+
+
+class FacilityFlagUpdateSpec(FacilityFlagBaseSpec):
+    @field_validator("flag")
+    @classmethod
+    def validate_flag_name(cls, flag_name):
+        FlagRegistry.validate_flag_name(FlagType.FACILITY, flag_name)
+        return flag_name
+
+
+class FacilityFlagReadSpec(FacilityFlagBaseSpec):
+    facility: dict
+
+    @classmethod
+    def perform_extra_serialization(cls, mapping, obj):
+        mapping["id"] = obj.external_id
+        mapping["facility"] = FacilityBareMinimumSpec.serialize(obj.facility).to_json()
+
+
+class FacilityFlagRetrieveSpec(FacilityFlagReadSpec):
+    pass

care/emr/resources/user_flag/spec.py

@@ -0,0 +1,50 @@
+from pydantic import UUID4, field_validator
+
+from care.emr.resources.base import EMRResource
+from care.emr.resources.user.spec import UserSpec
+from care.users.models import User, UserFlag
+from care.utils.registries.feature_flag import FlagRegistry, FlagType
+from care.utils.shortcuts import get_object_or_404
+
+
+class UserFlagBaseSpec(EMRResource):
+    __model__ = UserFlag
+    __exclude__ = ["user"]
+
+    id: UUID4 | None = None
+    flag: str
+
+
+class UserFlagCreateSpec(UserFlagBaseSpec):
+    user: UUID4
+
+    @field_validator("flag")
+    @classmethod
+    def validate_flag_name(cls, flag_name):
+        FlagRegistry.validate_flag_name(FlagType.USER, flag_name)
+        return flag_name
+
+    def perform_extra_deserialization(self, is_update, obj):
+        if not is_update:
+            obj.user = get_object_or_404(User, external_id=self.user)
+
+
+class UserFlagUpdateSpec(UserFlagBaseSpec):
+    @field_validator("flag")
+    @classmethod
+    def validate_flag_name(cls, flag_name):
+        FlagRegistry.validate_flag_name(FlagType.USER, flag_name)
+        return flag_name
+
+
+class UserFlagReadSpec(UserFlagBaseSpec):
+    user: dict
+
+    @classmethod
+    def perform_extra_serialization(cls, mapping, obj):
+        mapping["id"] = obj.external_id
+        mapping["user"] = UserSpec.serialize(obj.user).to_json()
+
+
+class UserFlagRetrieveSpec(UserFlagReadSpec):
+    pass