Skip to content

Commit 490401d

Browse files
rcmerlofarmerpaul
rcmerlo
and
farmerpaul
authored
feat: Enhances OneUp Health integration for document retrieval (M2-8883) (#1879)
* Enhances OneUp Health integration for document retrieval Improves the OneUp Health integration by adding functionality to download and store patient documents associated with EHR data. This change introduces the ability to fetch documents referenced in DocumentReference resources, store them in a zip file, and upload the zip to the EHR storage. It also introduces generic file uploading and listing capabilities in the EHR storage, along with helper methods. * Adds metadata to EHR answers Adds metadata to the EHR answers table to store information about the uploaded zip files, including their names and sizes. This allows for better tracking and management of the EHR data stored in the system. * Adds assertion to task ingest user data Adds an assertion to check that the result of the `ingest_user_data` function is not None. This ensures that the task completes successfully and returns a valid result. * Update src/apps/integrations/oneup_health/service/oneup_health.py Co-authored-by: Farmer Paul <paul.hh@metalab.com> * Refactors EHR metadata handling Updates EHR metadata to use a dedicated data model. This change introduces `EHRFileMetadata` and `EHRFileTypeEnum` to provide more structured and type-safe handling of EHR file metadata, replacing the previous use of dictionaries. This improves code clarity and maintainability. * Adds provider name to EHR document filename Ensures that the EHR document filename includes the healthcare provider's name (or ID if name unavailable) to improve identification and organization. * Adds buckets for answers and operations Extends MinIO bucket creation to include buckets for answers and operations, in addition to the existing media bucket. This allows for a more organized storage structure within MinIO and improves separation of concerns. * Adds a TODO item to optimize zip file creation. Adds a TODO item to address potential memory issues when creating zip files, especially when dealing with large documents or numerous files. * Refactors EHR file upload process Streamlines the EHR file upload process by moving the base path generation logic into the `EHRStorage.upload_file` method. This change improves code maintainability and reduces redundancy. --------- Co-authored-by: Farmer Paul <paul.hh@metalab.com>
1 parent db4583c commit 490401d

11 files changed

Lines changed: 237 additions & 27 deletions

File tree

‎compose/minio/create_bucket.sh‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,24 @@
77

88
/usr/bin/mc config host add local http://minio:9000 minioaccess miniosecret;
99
#/usr/bin/mc rm -r --force local/${CDN__BUCKET};
10+
#media bucket
1011
/usr/bin/mc mb -p local/${CDN__BUCKET};
1112
/usr/bin/mc policy set download local/${CDN__BUCKET};
1213
/usr/bin/mc policy set public local/${CDN__BUCKET};
1314
/usr/bin/mc anonymous set upload local/${CDN__BUCKET};
1415
/usr/bin/mc anonymous set download local/${CDN__BUCKET};
1516
/usr/bin/mc anonymous set public local/${CDN__BUCKET};
17+
#answer bucket
18+
/usr/bin/mc mb -p local/${CDN__BUCKET_ANSWER};
19+
/usr/bin/mc policy set download local/${CDN__BUCKET_ANSWER};
20+
/usr/bin/mc policy set public local/${CDN__BUCKET_ANSWER};
21+
/usr/bin/mc anonymous set upload local/${CDN__BUCKET_ANSWER};
22+
/usr/bin/mc anonymous set download local/${CDN__BUCKET_ANSWER};
23+
#operations bucket
24+
/usr/bin/mc mb -p local/${CDN__BUCKET_OPERATIONS};
25+
/usr/bin/mc policy set download local/${CDN__BUCKET_OPERATIONS};
26+
/usr/bin/mc policy set public local/${CDN__BUCKET_OPERATIONS};
27+
/usr/bin/mc anonymous set upload local/${CDN__BUCKET_OPERATIONS};
28+
/usr/bin/mc anonymous set download local/${CDN__BUCKET_OPERATIONS};
1629

17-
exit 0;
30+
exit 0;

‎docker-compose.yaml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,9 @@ services:
118118
image: minio/mc
119119
container_name: mindlogger_minio_mc
120120
environment:
121-
CDN__BUCKET: "${CDN__BUCKET:-media}"
121+
CDN__BUCKET: "${CDN__BUCKET:-cmi-media-local}"
122+
CDN__BUCKET_ANSWER: "${CDN__BUCKET_ANSWER:-cmi-answer-local}"
123+
CDN__BUCKET_OPERATIONS: "${CDN__BUCKET_OPERATIONS:-cmi-ops-local}"
122124
depends_on:
123125
- minio
124126
volumes:

‎src/apps/answers/db/schemas.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,3 +107,4 @@ class AnswerEHRSchema(Base):
107107
activity_id = Column(UUID(as_uuid=True), index=True)
108108
ehr_storage_uri = Column(Text())
109109
ehr_ingestion_status = Column(Text())
110+
meta = Column(JSONB())

‎src/apps/answers/domain/answers.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
from apps.activity_flows.domain.flow_full import FlowFull, FlowHistoryWithActivityFlat, FlowHistoryWithActivityFull
1919
from apps.answers.domain.answer_items import AnswerItem, ItemAnswerCreate
2020
from apps.applets.domain.base import AppletBaseInfo
21+
from apps.integrations.oneup_health.service.domain import EHRMetadata
2122
from apps.integrations.prolific.domain import ProlificParamsActivityAnswer
2223
from apps.shared.domain import InternalModel, PublicModel, Response
2324
from apps.shared.domain.custom_validations import datetime_from_ms
@@ -748,6 +749,7 @@ class AnswerEHR(InternalModel):
748749
ehr_ingestion_status: EHRIngestionStatus
749750
activity_id: uuid.UUID
750751
ehr_storage_uri: str | None
752+
meta: EHRMetadata | None
751753

752754

753755
class AnswerEHRFull(AnswerEHR):

‎src/apps/integrations/oneup_health/service/domain.py‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import uuid
22
from datetime import datetime
3+
from enum import StrEnum
34

45
from pydantic import Field
56

@@ -15,3 +16,19 @@ class EHRData(InternalModel):
1516
activity_id: uuid.UUID
1617
target_subject_id: uuid.UUID
1718
user_id: uuid.UUID
19+
20+
21+
class EHRFileTypeEnum(StrEnum):
22+
DOCS = "DOCS"
23+
EHR = "EHR"
24+
25+
26+
class EHRFileMetadata(InternalModel):
27+
name: str
28+
size: int
29+
type: EHRFileTypeEnum
30+
31+
32+
class EHRMetadata(InternalModel):
33+
zip_files: list[EHRFileMetadata]
34+
storage_path: str

‎src/apps/integrations/oneup_health/service/ehr_storage.py‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,19 @@ def ehr_zip_filename(data: EHRData) -> str:
2727
)
2828
return filename
2929

30+
@staticmethod
31+
def docs_zip_filename(data: EHRData) -> str:
32+
provider_name = (
33+
slugify(data.healthcare_provider_name, separator="-")
34+
if data.healthcare_provider_name
35+
else data.healthcare_provider_id
36+
)
37+
filename = (
38+
f"{data.target_subject_id}_{data.activity_id}_"
39+
f"{data.submit_id}_{data.date.strftime('%Y%m%d')}_{provider_name}_DOCS.zip"
40+
)
41+
return filename
42+
3043
def _get_storage_path(self, base_path: str, key: str) -> str:
3144
index = key.find(base_path)
3245
if index == -1: # substring not found
@@ -59,7 +72,18 @@ async def upload_resources(self, data: EHRData) -> tuple[str, str]:
5972

6073
return self._get_storage_path(base_path, key), filename
6174

62-
async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> str:
75+
async def upload_file(self, data: EHRData, filename: str, content: bytes) -> None:
76+
base_path = self._get_base_path(data)
77+
key = self._cdn_client.generate_key(FileScopeEnum.EHR, base_path, filename)
78+
79+
file_buffer = io.BytesIO(content)
80+
file_buffer.seek(0)
81+
82+
await self._cdn_client.upload(key, file_buffer)
83+
84+
file_buffer.close()
85+
86+
async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> tuple[str, int]:
6387
base_path = self._get_base_path(data)
6488
filename = EHRStorage.ehr_zip_filename(data)
6589
key = self._cdn_client.generate_key(FileScopeEnum.EHR, base_path, filename)
@@ -80,8 +104,9 @@ async def upload_ehr_zip(self, resources_files: list[str], data: EHRData) -> str
80104
file_buffer.close()
81105

82106
zip_buffer.seek(0)
107+
file_size = zip_buffer.getbuffer().nbytes
83108
await self._cdn_client.upload(key, zip_buffer)
84-
return key
109+
return filename, file_size
85110
finally:
86111
zip_buffer.close()
87112

‎src/apps/integrations/oneup_health/service/oneup_health.py‎

Lines changed: 111 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
1+
import base64
12
import hashlib
3+
import io
4+
import mimetypes
25
import uuid
6+
import zipfile
37
from datetime import datetime, timezone
48
from functools import reduce
59

610
import httpx
11+
from slugify import slugify
712

813
from apps.integrations.oneup_health.errors import (
914
OneUpHealthAPIError,
@@ -13,8 +18,8 @@
1318
OneUpHealthTokenExpiredError,
1419
OneUpHealthUserAlreadyExistsError,
1520
)
16-
from apps.integrations.oneup_health.service.domain import EHRData
17-
from apps.integrations.oneup_health.service.ehr_storage import create_ehr_storage
21+
from apps.integrations.oneup_health.service.domain import EHRData, EHRFileMetadata, EHRFileTypeEnum, EHRMetadata
22+
from apps.integrations.oneup_health.service.ehr_storage import EHRStorage, create_ehr_storage
1823
from apps.shared.exception import InternalServerError
1924
from config import settings
2025

@@ -480,6 +485,83 @@ async def _get_resources(self, entry_url: str, oneup_user_id: int):
480485

481486
return resources
482487

488+
def _get_document_references(self, resource_list) -> list[dict[str, str]]:
489+
"""Filter DocumentReference resources and extract their document URLs."""
490+
doc_refs = [res for res in resource_list if res.get("resourceType") == "DocumentReference"]
491+
documents = []
492+
for doc_ref in doc_refs:
493+
for content in doc_ref.get("content", []):
494+
attachment = content.get("attachment", {})
495+
title = attachment.get("title")
496+
url = attachment.get("url")
497+
if url:
498+
documents.append(dict(title=title, url=url))
499+
return documents
500+
501+
def _get_extension_from_content_type(self, content_type: str) -> str | None:
502+
"""Return the file extension (with dot) for a given MIME content type, or None if unknown."""
503+
# Handle common edge cases if needed
504+
if content_type == "application/xml":
505+
return ".xml"
506+
507+
if content_type == "application/json":
508+
return ".json"
509+
510+
return mimetypes.guess_extension(content_type)
511+
512+
async def _download_and_store_documents(
513+
self, ehr_storage, oneup_user_id: int, data: EHRData
514+
) -> tuple[str | None, int | None]:
515+
document_references = self._get_document_references(data.resources)
516+
517+
if len(document_references) == 0:
518+
logger.info(f"No documents found for activity_id {data.activity_id}, submit_id {data.submit_id}")
519+
return None, None
520+
521+
zip_filename = EHRStorage.docs_zip_filename(data)
522+
523+
# TODO: Optimize this function to avoid loading all files in memory when creating the zip file.
524+
# Current implementation loads all document content into memory before writing to the zip file,
525+
# which can cause memory issues with large documents or many documents.
526+
# Consider using a streaming approach or temporary files to reduce memory usage.
527+
zip_buffer = io.BytesIO()
528+
try:
529+
with zipfile.ZipFile(zip_buffer, "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
530+
for reference in document_references:
531+
url = reference["url"]
532+
try:
533+
document_meta = await self._client.get(url, headers={"x-oneup-user-id": str(oneup_user_id)})
534+
title = reference.get("title")
535+
last_updated_str = document_meta.get("meta", {}).get("lastUpdated")
536+
date = (
537+
datetime.fromisoformat(last_updated_str.replace("Z", "+00:00"))
538+
if last_updated_str
539+
else None
540+
)
541+
if date:
542+
date_string = date.strftime("%Y%m%d_%H%M%S")
543+
ext = self._get_extension_from_content_type(document_meta.get("contentType"))
544+
logger.info(f"Guessed extension: {ext} for mime type: {document_meta.get('contentType')}")
545+
file_name = f"{slugify(title) if title else url.split('/')[-1]}_{date_string}{ext}"
546+
data_b64: str = document_meta.get("data")
547+
content = base64.b64decode(data_b64)
548+
await ehr_storage.upload_file(data, file_name, content)
549+
zip_file.writestr(file_name, content)
550+
logger.info(f"Downloaded and stored document: {file_name}")
551+
552+
except OneUpHealthAPIError as ex:
553+
logger.error(f"Failed to download document: {url}: {ex}")
554+
continue
555+
556+
zip_buffer.seek(0)
557+
zip_size = zip_buffer.getbuffer().nbytes
558+
await ehr_storage.upload_file(data, zip_filename, zip_buffer.getvalue())
559+
560+
finally:
561+
zip_buffer.close()
562+
563+
return zip_filename, zip_size
564+
483565
async def retrieve_patient_data(
484566
self,
485567
session,
@@ -490,7 +572,7 @@ async def retrieve_patient_data(
490572
activity_id: uuid.UUID,
491573
oneup_user_id: int,
492574
healthcare_providers: list[dict[str, str]],
493-
) -> str | None:
575+
) -> EHRMetadata | None:
494576
"""
495577
Retrieve and store patient data for a subject.
496578
@@ -522,28 +604,28 @@ async def retrieve_patient_data(
522604
entries = result.get("entry", [])
523605
ehr_storage = await create_ehr_storage(session=session, applet_id=applet_id)
524606
resource_files = []
607+
zip_files = []
525608
for entry in entries:
526609
resource_url = entry.get("fullUrl")
527610
if resource_url:
528611
logger.info(f"Retrieving resources from {resource_url}")
529612
resources = await self._get_resources(f"{resource_url}/$everything?_count=100", oneup_user_id)
530613
if len(resources) > 0:
531-
# Get the healthcare provider name
532-
healthcare_provider_id = entry.get("resource", {}).get("id")
614+
# Get the healthcare provider meta data
533615
meta_source = entry.get("resource", {}).get("meta", {}).get("source")
534-
healthcare_provider_name = next(
616+
healthcare_provider = next(
535617
(
536-
healthcare_provider["name"]
618+
healthcare_provider
537619
for healthcare_provider in healthcare_providers
538620
if f"1up-external-system:{healthcare_provider['id']}" == meta_source
539621
),
540-
None,
622+
{},
541623
)
542624

543625
data = EHRData(
544626
resources=resources,
545-
healthcare_provider_id=healthcare_provider_id,
546-
healthcare_provider_name=healthcare_provider_name,
627+
healthcare_provider_id=healthcare_provider.get("id"),
628+
healthcare_provider_name=healthcare_provider.get("name"),
547629
date=datetime.now(timezone.utc),
548630
submit_id=submit_id,
549631
activity_id=activity_id,
@@ -555,8 +637,19 @@ async def retrieve_patient_data(
555637
resource_files.append(f"{storage_path}/{filename}")
556638
logger.info(
557639
f"Stored EHR data for healthcare provider "
558-
f"{healthcare_provider_name} in {storage_path}/{filename}"
640+
f"{healthcare_provider.get('name')} in {storage_path}/{filename}"
559641
)
642+
docs_zip_filename, docs_zip_size = await self._download_and_store_documents(
643+
ehr_storage, oneup_user_id, data
644+
)
645+
if docs_zip_filename:
646+
zip_files.append(
647+
EHRFileMetadata(name=docs_zip_filename, size=docs_zip_size, type=EHRFileTypeEnum.DOCS)
648+
)
649+
logger.info(
650+
f"Stored documents for healthcare provider {healthcare_provider.get('name')} "
651+
f"in {docs_zip_filename} size {docs_zip_size}"
652+
)
560653

561654
# Upload EHR zip with all resources
562655
data = EHRData(
@@ -566,8 +659,12 @@ async def retrieve_patient_data(
566659
target_subject_id=target_subject_id,
567660
user_id=user_id,
568661
)
569-
zip_file_path = await ehr_storage.upload_ehr_zip(resource_files, data)
662+
ehr_zip_filename, ehr_zip_size = await ehr_storage.upload_ehr_zip(resource_files, data)
663+
zip_files.append(EHRFileMetadata(name=ehr_zip_filename, size=ehr_zip_size, type=EHRFileTypeEnum.EHR))
570664

571-
logger.info(f"Stored EHR data in {zip_file_path}")
665+
logger.info(f"Stored EHR data in {ehr_zip_filename} size {ehr_zip_size}")
572666

573-
return storage_path
667+
return EHRMetadata(
668+
zip_files=zip_files,
669+
storage_path=storage_path,
670+
)

‎src/apps/integrations/oneup_health/service/task.py‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
from apps.answers.crud.answers import AnswersEHRCRUD
1111
from apps.answers.deps.preprocess_arbitrary import get_answer_session, preprocess_arbitrary_url
1212
from apps.answers.domain import AnswerEHR, EHRIngestionStatus
13+
from apps.integrations.oneup_health.service.domain import EHRMetadata
1314
from apps.integrations.oneup_health.service.oneup_health import OneupHealthService
1415
from apps.shared.exception import BaseError
1516
from broker import broker
@@ -57,7 +58,7 @@ async def _process_data_transfer(
5758
activity_id: uuid.UUID,
5859
oneup_user_id: int,
5960
start_date: datetime | None,
60-
) -> str | None:
61+
) -> EHRMetadata | None:
6162
"""
6263
Process the OneUp Health data transfer for a subject.
6364
@@ -208,7 +209,7 @@ async def task_ingest_user_data(
208209
Returns:
209210
list | None: List of retrieved resources if successful, None otherwise
210211
"""
211-
storage_path = None
212+
ehr_metadata = None
212213

213214
async with session_manager.get_session()() as session:
214215
info = await preprocess_arbitrary_url(applet_id=applet_id, session=session)
@@ -239,7 +240,7 @@ async def task_ingest_user_data(
239240

240241
# Process data transfer
241242

242-
storage_path = await _process_data_transfer(
243+
ehr_metadata = await _process_data_transfer(
243244
session=session,
244245
user_id=user_id,
245246
target_subject_id=target_subject_id,
@@ -249,7 +250,7 @@ async def task_ingest_user_data(
249250
oneup_user_id=oneup_user_id,
250251
start_date=start_date,
251252
)
252-
if storage_path is None:
253+
if ehr_metadata is None:
253254
logger.info(f"Data transfer not complete for OneUp Health user ID {oneup_user_id}")
254255
# Error retry count is reset to default 0 if we are not in an error state.
255256
to_reschedule = await _schedule_retry(
@@ -278,7 +279,8 @@ async def task_ingest_user_data(
278279
submit_id=submit_id,
279280
ehr_ingestion_status=EHRIngestionStatus.COMPLETED,
280281
activity_id=activity_id,
281-
ehr_storage_uri=storage_path,
282+
ehr_storage_uri=ehr_metadata.storage_path,
283+
meta=ehr_metadata,
282284
)
283285
)
284286
except (BaseError, httpx.RequestError) as e:
@@ -300,4 +302,4 @@ async def task_ingest_user_data(
300302
failed_attempts,
301303
)
302304

303-
return storage_path
305+
return ehr_metadata.storage_path if ehr_metadata is not None else None

0 commit comments

Comments
 (0)