this repo has no description
Observation Definition Done
+194
-9
+50
-6
care/emr/api/viewsets/observation_definition.py
+50
-6
care/emr/api/viewsets/observation_definition.py
···
1
+
from django_filters import rest_framework as filters
1
2
from rest_framework.exceptions import PermissionDenied
3
+
from rest_framework.generics import get_object_or_404
2
4
3
5
from care.emr.api.viewsets.base import (
4
6
EMRBaseViewSet,
···
13
15
ObservationDefinitionCreateSpec,
14
16
ObservationDefinitionReadSpec,
15
17
)
18
+
from care.facility.models import Facility
19
+
from care.security.authorization import AuthorizationController
20
+
21
+
22
+
class ObservationDefinitionFilters(filters.FilterSet):
23
+
facility = filters.UUIDFilter(field_name="facility__external_id")
16
24
17
25
18
26
class ObservationViewSet(
···
22
30
pydantic_model = ObservationDefinitionCreateSpec
23
31
pydantic_update_model = BaseObservationDefinitionSpec
24
32
pydantic_read_model = ObservationDefinitionReadSpec
33
+
filterset_class = ObservationDefinitionFilters
34
+
filter_backends = [filters.DjangoFilterBackend]
25
35
26
36
def authorize_create(self, instance):
27
-
if not self.request.user.is_superuser:
37
+
"""
38
+
Only superusers can create observation definitions that are not facility-specific.
39
+
The user must have permission to create the observation definition in the facility.
40
+
"""
41
+
if not instance.facility and not self.request.user.is_superuser:
42
+
raise PermissionDenied("Access Denied to Observation Definition")
43
+
if instance.facility and not AuthorizationController.call(
44
+
"can_write_facility_observation_definition",
45
+
self.request.user,
46
+
instance.facility,
47
+
):
28
48
raise PermissionDenied("Access Denied to Observation Definition")
29
49
30
-
def authorize_update(self, instance):
31
-
if not self.request.user.is_superuser:
50
+
def authorize_update(self, request_obj, model_instance):
51
+
"""
52
+
Only superusers can update observation definitions that are not facility-specific.
53
+
The user must have permission to update the observation definition in the facility.
54
+
"""
55
+
if not model_instance.facility and not self.request.user.is_superuser:
56
+
raise PermissionDenied("Access Denied to Observation Definition")
57
+
58
+
if model_instance.facility and not AuthorizationController.call(
59
+
"can_write_facility_observation_definition",
60
+
self.request.user,
61
+
model_instance.facility,
62
+
):
32
63
raise PermissionDenied("Access Denied to Observation Definition")
33
64
34
65
def get_queryset(self):
35
-
if self.request.user.is_superuser:
36
-
return self.database_model.objects.all()
37
-
return self.database_model.objects.none()
66
+
"""
67
+
If no facility filters are applied, all objects must be returned without a facility filter.
68
+
If facility filter is applied, check for read permission and return all inside facility.
69
+
"""
70
+
base_queryset = self.database_model.objects.all()
71
+
if "facility" in self.request.GET:
72
+
facility_id = self.request.GET["facility"]
73
+
facility_obj = get_object_or_404(Facility, external_id=facility_id)
74
+
if not AuthorizationController.call(
75
+
"can_list_facility_observation_definition",
76
+
self.request.user,
77
+
facility_obj,
78
+
):
79
+
raise PermissionDenied("Access Denied to Observation Definition")
80
+
return base_queryset.filter(facility=facility_obj)
81
+
return base_queryset.filter(facility__is_null=True)
+49
care/emr/migrations/0022_observationdefinition.py
+49
care/emr/migrations/0022_observationdefinition.py
···
1
+
# Generated by Django 5.1.4 on 2025-04-07 07:16
2
+
3
+
import django.db.models.deletion
4
+
import uuid
5
+
from django.conf import settings
6
+
from django.db import migrations, models
7
+
8
+
9
+
class Migration(migrations.Migration):
10
+
11
+
dependencies = [
12
+
('emr', '0021_metaartifact'),
13
+
('facility', '0476_facility_default_internal_organization_and_more'),
14
+
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
15
+
]
16
+
17
+
operations = [
18
+
migrations.CreateModel(
19
+
name='ObservationDefinition',
20
+
fields=[
21
+
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
22
+
('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
23
+
('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
24
+
('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
25
+
('deleted', models.BooleanField(db_index=True, default=False)),
26
+
('history', models.JSONField(default=dict)),
27
+
('meta', models.JSONField(default=dict)),
28
+
('version', models.IntegerField(default=1)),
29
+
('slug', models.CharField(max_length=255)),
30
+
('name', models.CharField(max_length=1024)),
31
+
('status', models.CharField(max_length=255)),
32
+
('description', models.TextField()),
33
+
('derived_from_uri', models.TextField()),
34
+
('category', models.JSONField()),
35
+
('code', models.JSONField()),
36
+
('permitted_data_type', models.JSONField()),
37
+
('body_site', models.JSONField()),
38
+
('method', models.JSONField()),
39
+
('permitted_unit', models.JSONField()),
40
+
('component', models.JSONField()),
41
+
('created_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_created_by', to=settings.AUTH_USER_MODEL)),
42
+
('facility', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.PROTECT, to='facility.facility')),
43
+
('updated_by', models.ForeignKey(blank=True, default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(app_label)s_%(class)s_updated_by', to=settings.AUTH_USER_MODEL)),
44
+
],
45
+
options={
46
+
'abstract': False,
47
+
},
48
+
),
49
+
]
+2
-1
care/emr/models/observation_definition.py
+2
-1
care/emr/models/observation_definition.py
···
12
12
blank=True,
13
13
)
14
14
version = models.IntegerField(default=1)
15
-
title = models.TextField()
15
+
slug = models.CharField(max_length=255)
16
+
name = models.CharField(max_length=1024)
16
17
status = models.CharField(max_length=255)
17
18
description = models.TextField()
18
19
derived_from_uri = models.TextField()
+26
-2
care/emr/resources/observation_definition/spec.py
+26
-2
care/emr/resources/observation_definition/spec.py
···
3
3
from care.emr.models.observation_definition import ObservationDefinition
4
4
from care.emr.resources.base import EMRResource
5
5
from care.emr.resources.common import Coding
6
+
from care.emr.resources.facility.spec import FacilityBareMinimumSpec
6
7
from care.emr.resources.observation.valueset import (
7
8
CARE_BODY_SITE_VALUESET,
8
9
CARE_OBSERVATION_COLLECTION_METHOD,
9
10
)
10
11
from care.emr.resources.questionnaire.spec import QuestionType
12
+
from care.facility.models import Facility
11
13
12
14
13
15
class ObservationDefinitionComponentSpec(BaseModel):
···
28
30
__exclude__ = ["facility"]
29
31
30
32
id: str
31
-
title: str
33
+
slug: str
34
+
name: str
32
35
status: str
33
36
description: str
34
37
category: Coding | None = None
···
44
47
json_schema_extra={"slug": CARE_OBSERVATION_COLLECTION_METHOD.slug},
45
48
)
46
49
permitted_unit: Coding
50
+
derived_from_uri: str | None = None
47
51
48
52
@field_validator("permitted_data_type")
49
53
@classmethod
···
55
59
56
60
class ObservationDefinitionCreateSpec(BaseObservationDefinitionSpec):
57
61
facility: UUID4 | None = None
58
-
derivedFromUri: str
62
+
63
+
@field_validator("facility")
64
+
@classmethod
65
+
def validate_facility_exists(cls, facility):
66
+
if not Facility.objects.filter(external_id=facility).exists():
67
+
err = "Facility not found"
68
+
raise ValueError(err)
69
+
return facility
70
+
71
+
def perform_extra_deserialization(self, is_update, obj):
72
+
if self.facility:
73
+
obj.facility = Facility.objects.get(external_id=self.facility)
59
74
60
75
61
76
class ObservationDefinitionReadSpec(BaseObservationDefinitionSpec):
62
77
version: int | None = None
78
+
facility: dict | None = None
79
+
80
+
@classmethod
81
+
def perform_extra_serialization(cls, mapping, obj):
82
+
mapping["id"] = obj.external_id
83
+
if obj.facility:
84
+
mapping["facility"] = FacilityBareMinimumSpec.serialize(
85
+
obj.faciltiy
86
+
).to_json()
+4
care/security/permissions/base.py
+4
care/security/permissions/base.py
···
4
4
FacilityOrganizationPermissions,
5
5
)
6
6
from care.security.permissions.location import FacilityLocationPermissions
7
+
from care.security.permissions.observation_definition import (
8
+
ObservationDefinitionPermissions,
9
+
)
7
10
from care.security.permissions.organization import OrganizationPermissions
8
11
from care.security.permissions.patient import PatientPermissions
9
12
from care.security.permissions.questionnaire import QuestionnairePermissions
···
34
37
UserPermissions,
35
38
UserSchedulePermissions,
36
39
FacilityLocationPermissions,
40
+
ObservationDefinitionPermissions,
37
41
]
38
42
39
43
cache = {}
+35
care/security/permissions/observation_definition.py
+35
care/security/permissions/observation_definition.py
···
1
+
import enum
2
+
3
+
from care.security.permissions.constants import Permission, PermissionContext
4
+
from care.security.roles.role import (
5
+
ADMIN_ROLE,
6
+
DOCTOR_ROLE,
7
+
FACILITY_ADMIN_ROLE,
8
+
GEO_ADMIN,
9
+
NURSE_ROLE,
10
+
STAFF_ROLE,
11
+
VOLUNTEER_ROLE,
12
+
)
13
+
14
+
15
+
class ObservationDefinitionPermissions(enum.Enum):
16
+
can_write_observation_definition = Permission(
17
+
"Can Create Observation Definition on Facility",
18
+
"",
19
+
PermissionContext.FACILITY,
20
+
[FACILITY_ADMIN_ROLE, ADMIN_ROLE],
21
+
)
22
+
can_read_observation_definition = Permission(
23
+
"Can Read Observation Definition",
24
+
"",
25
+
PermissionContext.FACILITY,
26
+
[
27
+
FACILITY_ADMIN_ROLE,
28
+
GEO_ADMIN,
29
+
ADMIN_ROLE,
30
+
STAFF_ROLE,
31
+
DOCTOR_ROLE,
32
+
NURSE_ROLE,
33
+
VOLUNTEER_ROLE,
34
+
],
35
+
)