Merge pull request #111 from Lightmatter/develop

v0.1.5

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "welkin"
-version = "0.1.4"
+version = "0.1.5"
 description = "Python Welkin Health API Wrapper."
 authors = ["Sam Morgan <[email protected]>"]
 license = "GPL-3.0-or-later"

test/test_export.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.mark.vcr
+def test_cdt_records_export_read(client, vcr_cassette):
+    export = client.CDTRecordsExport()
+
+    exports = list(export.get(paginate=True, size=1000))
+
+    assert len(exports) > 1
+    assert len(vcr_cassette) == 6

welkin/client.py

@@ -82,6 +82,7 @@ class Client(Session):
     CarePlan = models.CarePlan
     CarePlanOverview = models.CarePlanOverview
     CDT = models.CDT
+    CDTRecordsExport = models.CDTRecordsExport
     CDTs = models.CDTs
     Chat = models.Chat
     Chats = models.Chats
@@ -96,11 +97,11 @@ class Client(Session):
     Encounters = models.Encounters
     Formation = models.Formation
     Patient = models.Patient
-    Patients = models.Patients
     PatientProgram = models.PatientProgram
+    PatientPrograms = models.PatientPrograms
+    Patients = models.Patients
     ProgramPhase = models.ProgramPhase
     ProgramPhases = models.ProgramPhases
-    PatientPrograms = models.PatientPrograms
     Schedules = models.Schedules
     SearchChats = models.SearchChats
     SMS = models.SMS

welkin/models/__init__.py

@@ -17,6 +17,7 @@
 )
 from welkin.models.email import Email, Emails
 from welkin.models.encounter import Encounter, EncounterDisposition, Encounters
+from welkin.models.export import CDTRecordsExport
 from welkin.models.formation import Formation
 from welkin.models.patient import Patient, Patients
 from welkin.models.program import (
@@ -39,6 +40,7 @@
     "CarePlan",
     "CarePlanOverview",
     "CDT",
+    "CDTRecordsExport",
     "CDTs",
     "Chat",
     "Chats",
@@ -53,11 +55,11 @@
     "Encounters",
     "Formation",
     "Patient",
-    "Patients",
     "PatientProgram",
+    "PatientPrograms",
+    "Patients",
     "ProgramPhase",
     "ProgramPhases",
-    "PatientPrograms",
     "Schedules",
     "SearchChats",
     "SMS",

welkin/models/export.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from welkin.models.base import Collection, Resource
+from welkin.models.encounter import Encounters
+from welkin.pagination import CursorIterator
+
+
+class CDTRecordExport(Resource):
+    subresources = (Encounters,)
+
+    def __str__(self):
+        try:
+            return f"{self.firstName} {self.lastName}"
+        except AttributeError:
+            return self.username
+
+
+class CDTRecordsExport(Collection):
+    resource = CDTRecordExport
+    iterator = CursorIterator
+
+    def get(self, *args, **kwargs):
+        return super().get(f"{self._client.instance}/export/CDT_RECORD", *args, **kwargs)

welkin/pagination.py

@@ -194,3 +194,19 @@ def _pre_request(self):
     def _post_request(self, meta):
         self.page_token = meta["nextPageToken"]
         self.last = not meta.get("nextPageToken")
+
+
+class CursorIterator(PageIterator):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.meta_key = "cursor"
+        self.meta_dict = {
+            "start": None,
+            "end": None,
+            "pageSize": self.size,
+        }
+        self.kwargs.setdefault("params", {}).update(pageSize=self.size)
+
+    def _post_request(self, meta):
+        self.kwargs.setdefault("params", {}).update(start=meta.get("end"))
+        self.last = not meta.get("hasNext")