Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 108 additions & 0 deletions care/emr/api/viewsets/facility_flag.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
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():
super().perform_create(instance)
FlagRegistry.register(FlagType.FACILITY, instance.flag)

def perform_destroy(self, instance):
with transaction.atomic():
flag_name = instance.flag
super().perform_destroy(instance)

still_used = FacilityFlag.objects.filter(
flag=flag_name, deleted=False
).exists()

if not still_used:
FlagRegistry.unregister(FlagType.FACILITY, flag_name)

@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
)
89 changes: 89 additions & 0 deletions care/emr/api/viewsets/user_flag.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
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():
super().perform_create(instance)
FlagRegistry.register(FlagType.USER, instance.flag)

def perform_destroy(self, instance):
with transaction.atomic():
flag_name = instance.flag
super().perform_destroy(instance)

still_used = UserFlag.objects.filter(flag=flag_name, deleted=False).exists()

if not still_used:
FlagRegistry.unregister(FlagType.USER, flag_name)

@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
)
Empty file.
51 changes: 51 additions & 0 deletions care/emr/resources/facility_flag/spec.py
Original file line number Diff line number Diff line change
@@ -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
Empty file.
50 changes: 50 additions & 0 deletions care/emr/resources/user_flag/spec.py
Original file line number Diff line number Diff line change
@@ -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
Loading