Added TODO items

care/emr/api/viewsets/location.py

@@ -5,8 +5,20 @@
 from rest_framework.generics import get_object_or_404
 from rest_framework.response import Response
 
-from care.emr.api.viewsets.base import EMRModelViewSet
-from care.emr.models import Encounter, FacilityLocation, FacilityLocationOrganization
+from care.emr.api.viewsets.base import (
+    EMRBaseViewSet,
+    EMRCreateMixin,
+    EMRListMixin,
+    EMRModelViewSet,
+    EMRRetrieveMixin,
+    EMRUpdateMixin,
+)
+from care.emr.models import (
+    Encounter,
+    FacilityLocation,
+    FacilityLocationEncounter,
+    FacilityLocationOrganization,
+)
 from care.emr.models.organization import FacilityOrganization, FacilityOrganizationUser
 from care.emr.resources.facility_organization.spec import FacilityOrganizationReadSpec
 from care.emr.resources.location.spec import (
@@ -190,3 +202,47 @@ def associate_encounter(self, request, *args, **kwargs):
         ):
             raise PermissionDenied("You do not have permission to update encounter")
         # TODO, Association models yet to be built
+
+
+class FacilityLocationEncounterFilter(filters.FilterSet):
+    encounter = filters.UUIDFilter(field_name="encounter__external_id")
+
+
+class FacilityLocationEncounterViewSet(
+    EMRCreateMixin, EMRRetrieveMixin, EMRUpdateMixin, EMRListMixin, EMRBaseViewSet
+):
+    """
+    TODO Authz for encounter creates
+    TODO Update encounter model when creates are done
+    TODO check for conflicts in datetime
+    TODO Add locks when a bed is occupied
+    TODO detect end dates added to encounter and remove association with encounter
+    TODO encounter lists must also check the location based access now
+    """
+
+    database_model = FacilityLocationEncounter
+    pydantic_model = None
+    pydantic_read_model = None
+    pydantic_retrieve_model = None
+    pydantic_update_model = None
+    filterset_class = FacilityLocationEncounterFilter
+    filter_backends = [filters.DjangoFilterBackend]
+
+    def get_facility_obj(self):
+        return get_object_or_404(
+            Facility, external_id=self.kwargs["facility_external_id"]
+        )
+
+    def get_location_obj(self):
+        return get_object_or_404(
+            FacilityLocation, external_id=self.kwargs["location_external_id"]
+        )
+
+    def get_queryset(self):
+        location = self.get_location_obj()
+        facility = self.get_facility_obj()
+        if not AuthorizationController.call(
+            "can_list_facility_location_obj", self.request.user, facility, location
+        ):
+            raise PermissionDenied("You do not have permission to given location")
+        return FacilityLocationEncounter.objects.filter(location=location)

care/emr/migrations/0014_encounter_current_location_facilitylocationencounter.py

@@ -0,0 +1,43 @@
+# Generated by Django 5.1.4 on 2025-01-26 21:03
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('emr', '0013_facilitylocation_facilitylocationorganization'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='encounter',
+            name='current_location',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='emr.facilitylocation'),
+        ),
+        migrations.CreateModel(
+            name='FacilityLocationEncounter',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('history', models.JSONField(default=dict)),
+                ('meta', models.JSONField(default=dict)),
+                ('start_datetime', models.DateTimeField()),
+                ('end_datetime', models.DateTimeField(blank=True, default=None, null=True)),
+                ('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)),
+                ('encounter', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.encounter')),
+                ('location', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='emr.facilitylocation')),
+                ('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)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

care/emr/models/encounter.py

@@ -21,6 +21,10 @@ class Encounter(EMRBaseModel):
     # Organization fields
     facility_organization_cache = ArrayField(models.IntegerField(), default=list)
 
+    current_location = models.ForeignKey(
+        "emr.FacilityLocation", on_delete=models.SET_NULL, null=True, blank=True
+    )  # Cached field, used for easier querying
+
     def sync_organization_cache(self):
         orgs = set()
         for encounter_organization in EncounterOrganization.objects.filter(

care/emr/models/location.py

@@ -4,7 +4,7 @@
 from django.db import models
 from django.utils import timezone
 
-from care.emr.models import EMRBaseModel, FacilityOrganization
+from care.emr.models import EMRBaseModel, Encounter, FacilityOrganization
 from config.celery_app import app
 
 
@@ -131,6 +131,17 @@ def save(self, *args, **kwargs):
         handle_cascade.delay(self.location.id)
 
 
+class FacilityLocationEncounter(EMRBaseModel):
+    """
+    This relation denotes how a bed was associated to an encounter
+    """
+
+    location = models.ForeignKey(FacilityLocation, on_delete=models.CASCADE)
+    encounter = models.ForeignKey(Encounter, on_delete=models.CASCADE)
+    start_datetime = models.DateTimeField()
+    end_datetime = models.DateTimeField(default=None, null=True, blank=True)
+
+
 @app.task
 def handle_cascade(base_location):
     """

care/emr/resources/base.py

@@ -145,6 +145,15 @@ def as_questionnaire(cls, parent_classes=None):  # noqa PLR0912
     def to_json(self):
         return self.model_dump(mode="json", exclude=["meta"])
 
+    @classmethod
+    def serialize_audit_users(cls, mapping, obj):
+        from care.emr.resources.user.spec import UserSpec
+
+        if obj.created_by:
+            mapping["created_by"] = UserSpec.serialize(obj.created_by)
+        if obj.updated_by:
+            mapping["updated_by"] = UserSpec.serialize(obj.updated_by)
+
 
 PhoneNumber = Annotated[
     Union[str, phonenumbers.PhoneNumber()],  # noqa: UP007

care/emr/resources/location/spec.py

@@ -1,7 +1,9 @@
+import datetime
 from enum import Enum
 
 from pydantic import UUID4, model_validator
 
+from care.emr.models import Encounter, FacilityLocationEncounter
 from care.emr.models.location import FacilityLocation
 from care.emr.resources.base import EMRResource
 from care.emr.resources.common import Coding
@@ -101,10 +103,53 @@ def perform_extra_serialization(cls, mapping, obj):
 
 
 class FacilityLocationRetrieveSpec(FacilityLocationListSpec):
+    created_by: dict | None = None
+    updated_by: dict | None = None
+
     @classmethod
     def perform_extra_serialization(cls, mapping, obj):
         super().perform_extra_serialization(mapping, obj)
         if obj.created_by:
             mapping["created_by"] = UserSpec.serialize(obj.created_by)
         if obj.updated_by:
             mapping["updated_by"] = UserSpec.serialize(obj.updated_by)
+
+
+class FacilityLocationEncounterBaseSpec(EMRResource):
+    __model__ = FacilityLocationEncounter
+    __exclude__ = ["encounter", "location"]
+
+    id: UUID4 | None = None
+
+
+class FacilityLocationEncounterCreateSpec(FacilityLocationEncounterBaseSpec):
+    encounter: UUID4
+    start_datetime: datetime.datetime
+
+    @model_validator(mode="after")
+    def validate_encounter(self):
+        if not Encounter.objects.filter(external_id=self.encounter).exists():
+            err = "Encounter not found"
+            raise ValueError(err)
+        return self
+
+    def perform_extra_deserialization(self, is_update, obj):
+        obj.encounter = Encounter.objects.get(external_id=self.encounter)
+
+
+class FacilityLocationEncounterUpdateSpec(FacilityLocationEncounterBaseSpec):
+    start_datetime: datetime.datetime
+    end_datetime: datetime.datetime
+
+
+class FacilityLocationEncounterReadSpec(FacilityLocationEncounterBaseSpec):
+    encounter: UUID4
+    start_datetime: datetime.datetime
+    end_datetime: datetime.datetime | None = None
+
+    created_by: dict | None = None
+    updated_by: dict | None = None
+
+    @classmethod
+    def perform_extra_serialization(cls, mapping, obj):
+        cls.serialize_audit_users(mapping, obj)