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
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -90,8 +90,8 @@ format:
echo $$CHANGED_FILES; \
if [ -z "$$CHANGED_FILES" ]; then echo "No changed files to format"; exit 0; fi; \
fi; \
$(VENV_PATH_PREFIX)/bin/python3 -m black $$CHANGED_FILES; \
$(VENV_PATH_PREFIX)/bin/ruff check $$CHANGED_FILES --fix; \
$(VENV_PATH_PREFIX)/bin/python3 -m black $$CHANGED_FILES; \
$(VENV_PATH_PREFIX)/bin/python3 -m isort --profile black $$CHANGED_FILES

sort-requirements:
Expand Down
93 changes: 59 additions & 34 deletions lambdas/models/fhir/R4/fhir_document_reference.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
CodeableConcept,
Coding,
Extension,
Identifier,
Meta,
Period,
Reference,
Expand Down Expand Up @@ -64,7 +65,7 @@ class ContentStabilityExtensionValueCodeableConcept(CodeableConcept):
"""CodeableConcept for content stability."""

coding: List[ContentStabilityExtensionCoding] = Field(
default_factory=lambda: [ContentStabilityExtensionCoding()]
default_factory=lambda: [ContentStabilityExtensionCoding()],
)


Expand All @@ -73,7 +74,7 @@ class ContentStabilityExtension(Extension):

url: Literal[CONTENT_STABILITY_URL] = CONTENT_STABILITY_URL
valueCodeableConcept: ContentStabilityExtensionValueCodeableConcept = Field(
default_factory=ContentStabilityExtensionValueCodeableConcept
default_factory=ContentStabilityExtensionValueCodeableConcept,
)


Expand Down Expand Up @@ -167,7 +168,7 @@ def _create_identifier(self, system_suffix: str, value: str) -> Dict[str, Any]:
"identifier": {
"system": f"{FHIR_BASE_URL}/{system_suffix}",
"value": value,
}
},
}

def _create_snomed_coding(self, snomed_code: SnomedCode) -> List[Dict[str, str]]:
Expand All @@ -184,7 +185,7 @@ def _create_snomed_coding(self, snomed_code: SnomedCode) -> List[Dict[str, str]]
"system": SNOMED_URL,
"code": snomed_code.code,
"display": snomed_code.display_name,
}
},
]

def create_nrl_fhir_document_reference_object(self) -> DocumentReference:
Expand All @@ -204,25 +205,27 @@ def create_nrl_fhir_document_reference_object(self) -> DocumentReference:
subject=Reference(**self._create_identifier("nhs-number", self.nhs_number)),
content=[DocumentReferenceContent(attachment=self.attachment)],
custodian=Reference(
**self._create_identifier("ods-organization-code", self.custodian)
**self._create_identifier("ods-organization-code", self.custodian),
),
type=CodeableConcept(
coding=self._create_snomed_coding(self.snomed_code_doc_type)
coding=self._create_snomed_coding(self.snomed_code_doc_type),
),
category=[
CodeableConcept(
coding=self._create_snomed_coding(self.snomed_code_category)
)
coding=self._create_snomed_coding(self.snomed_code_category),
),
],
author=[
Reference(
**self._create_identifier("ods-organization-code", self.custodian)
)
**self._create_identifier("ods-organization-code", self.custodian),
),
],
context=DocumentReferenceContext(
practiceSetting=CodeableConcept(
coding=self._create_snomed_coding(self.snomed_code_practice_setting)
)
coding=self._create_snomed_coding(
self.snomed_code_practice_setting,
),
),
),
)

Expand All @@ -233,58 +236,80 @@ def create_nrl_fhir_document_reference_object(self) -> DocumentReference:
return fhir_document_ref

def create_fhir_document_reference_object(
self, document: NdrDocumentReference
self,
document: NdrDocumentReference,
) -> DocumentReference:
"""Create a FHIR DocumentReference .

Returns:
DocumentReference: A FHIR DocumentReference resource
"""
if document.custodian in PatientOdsInactiveStatus.list():
document.custodian = PCSE_ODS_CODE
author_ods = None
custodian_ods = None
if self.snomed_code_doc_type.code != SnomedCodes.PATIENT_DATA.value.code:
if document.custodian in PatientOdsInactiveStatus.list():
document.custodian = PCSE_ODS_CODE

author_ods = document.author or self.custodian
custodian_ods = document.custodian or self.custodian
else:
if document.author:
author_ods = document.author
if document.custodian:
custodian_ods = document.custodian

return DocumentReference(
fhir_doc_ref = DocumentReference(
resourceType="DocumentReference",
id=f"{self.snomed_code_doc_type.code}~{document.id}",
docStatus=document.doc_status,
type=CodeableConcept(
coding=self._create_snomed_coding(self.snomed_code_doc_type)
coding=self._create_snomed_coding(self.snomed_code_doc_type),
),
subject=Reference(**self._create_identifier("nhs-number", self.nhs_number)),
content=[DocumentReferenceContent(attachment=self.attachment)],
date=document.created,
author=[
Reference(
**self._create_identifier(
"ods-organization-code", document.author or self.custodian
)
)
],
custodian=Reference(
**self._create_identifier(
"ods-organization-code", document.custodian or self.custodian
)
),
meta=Meta(versionId=document.version),
)

if author_ods:
fhir_doc_ref.author = [
Reference(
identifier=Identifier(
system=f"{FHIR_BASE_URL}/ods-organization-code",
value=author_ods,
),
),
]
if custodian_ods:
fhir_doc_ref.custodian = Reference(
identifier=Identifier(
system=f"{FHIR_BASE_URL}/ods-organization-code",
value=custodian_ods,
),
)

return fhir_doc_ref

def create_fhir_document_reference_object_basic(
self, original_id: str, original_version
self,
original_id: str,
original_version,
) -> DocumentReference:
return DocumentReference(
resourceType="DocumentReference",
id=f"{original_id}",
type=CodeableConcept(
coding=self._create_snomed_coding(self.snomed_code_doc_type)
coding=self._create_snomed_coding(self.snomed_code_doc_type),
),
subject=Reference(**self._create_identifier("nhs-number", self.nhs_number)),
content=[DocumentReferenceContent(attachment=self.attachment)],
author=[
Reference(
**self._create_identifier(
"ods-organization-code", self.author or self.custodian
)
)
"ods-organization-code",
self.author or self.custodian,
),
),
],
meta=Meta(versionId=original_version),
)
Loading