[CDAPI-150]: Added validation ensuring an empty string cannot be prov…

.vscode/settings.json

@@ -25,6 +25,19 @@
   "python.telemetry.enable": false,
   "debugpy.telemetry.enable": false,
   "python.experiments.enabled": false,
+  // Exclude generated files from analysis
+  "python.analysis.exclude": [
+    "**/target",
+    "**/__pycache__",
+    "**/dist",
+    "**/build",
+    "**/.venv",
+    "**/.mypy_cache",
+    "**/.pytest_cache",
+    "**/.ruff_cache",
+    "**/node_modules",
+    "**/.*"
+  ],
   // Enabling Ruff formatter for python files.
   "[python]": {
     "editor.formatOnSave": true,
@@ -64,5 +77,6 @@
     "projectKey": "NHSDigital_clinical-data-pathology-api"
   },
   // Disabling automatic port forwarding as the devcontainer should already have access to any required ports.
-  "remote.autoForwardPorts": false
+  "remote.autoForwardPorts": false,
+  "python-envs.defaultEnvManager": "ms-python.python:system"
 }

bruno/APIM/Post_Document_Bundle_via_APIM.bru

@@ -23,13 +23,51 @@ body:json {
         "fullUrl": "composition",
         "resource": {
           "resourceType": "Composition",
+          "extension": [
+            {
+              "url": "http://hl7.eu/fhir/StructureDefinition/composition-basedOn-order-or-requisition",
+              "valueReference": {
+                "reference": "servicerequest"
+              }
+            }
+          ],
           "subject": {
             "identifier": {
               "system": "https://fhir.nhs.uk/Id/nhs-number",
               "value": "test-nhs-number"
             }
           }
         }
+      },
+      {
+        "fullUrl": "servicerequest",
+        "resource": {
+          "resourceType": "ServiceRequest",
+          "requester": {
+            "reference": "practitionerrole"
+          }
+        }
+      },
+      {
+        "fullUrl": "practitionerrole",
+        "resource": {
+          "resourceType": "PractitionerRole",
+          "organization": {
+            "reference": "organization"
+          }
+        }
+      },
+      {
+        "fullUrl": "organization",
+        "resource": {
+          "resourceType": "Organization",
+          "identifier": [
+            {
+              "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+              "value": "testOrg"
+            }
+          ]
+        }
       }
     ]
   }

pathology-api/conftest.py

@@ -0,0 +1,38 @@
+from collections.abc import Callable
+
+import pytest
+from pathology_api.fhir.r4.elements import (
+    LiteralReference,
+    LogicalReference,
+    OrganizationIdentifier,
+    PatientIdentifier,
+    ReferenceExtension,
+)
+from pathology_api.fhir.r4.resources import (
+    Bundle,
+    Composition,
+    Organization,
+)
+from pathology_api.test_utils import BundleBuilder
+
+
+@pytest.fixture(scope="session")
+def build_valid_test_result() -> Callable[[str, str], Bundle]:
+    def builder_function(patient: str, ods_code: str) -> Bundle:
+        return BundleBuilder.with_defaults(
+            composition_func=lambda service_request_url: Composition.create(
+                subject=LogicalReference(PatientIdentifier.from_nhs_number(patient)),
+                extension=[
+                    # Using HTTP to match profile required by implementation guide.
+                    ReferenceExtension(
+                        url="http://hl7.eu/fhir/StructureDefinition/composition-basedOn-order-or-requisition",  # noqa: S5332
+                        valueReference=LiteralReference(service_request_url),
+                    )
+                ],
+            ),
+            organisation_func=lambda: Organization.create(
+                identifier=[OrganizationIdentifier.from_ods_code(ods_code)]
+            ),
+        ).build()
+
+    return builder_function

pathology-api/src/pathology_api/fhir/r4/elements.py

@@ -2,9 +2,15 @@
 import uuid
 from abc import ABC
 from dataclasses import dataclass
-from typing import Annotated, ClassVar
+from typing import Annotated, Any, ClassVar
 
-from pydantic import Field, model_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    ValidatorFunctionWrapHandler,
+    model_validator,
+)
 
 from pathology_api.exception import ValidationError
 
@@ -36,41 +42,80 @@ def with_last_updated(cls, last_updated: datetime.datetime | None = None) -> "Me
         )
 
 
-@dataclass(frozen=True)
-class Identifier(ABC):
+class Identifier(ABC, BaseModel):
     """
     A FHIR R4 Identifier element. See https://hl7.org/fhir/R4/datatypes.html#Identifier.
     Attributes:
         system: The namespace for the identifier value.
         value: The value that is unique within the system.
     """
 
-    _expected_system: ClassVar[str] = "__unknown__"
+    __system_types: ClassVar[dict[str, type["Identifier"]]] = {}
+
+    _validate_system: ClassVar[bool] = True
+    expected_system: ClassVar[str] = "__unknown__"
 
-    system: str
-    value: str
+    system: str = Field(..., frozen=True)
+    value: str = Field(..., frozen=True)
 
     @model_validator(mode="after")
     def validate_system(self) -> "Identifier":
-        if self.system != self._expected_system:
+        if self._validate_system and self.system != self.expected_system:
             raise ValidationError(
                 f"Identifier system '{self.system}' does not match expected "
-                f"system '{self._expected_system}'."
+                f"system '{self.expected_system}'."
             )
         return self
 
     @classmethod
-    def __init_subclass__(cls, expected_system: str) -> None:
-        cls._expected_system = expected_system
+    def __init_subclass__(
+        cls, expected_system: str = "__unknown__", validate_system: bool = True
+    ) -> None:
+        cls.expected_system = expected_system
+        cls._validate_system = validate_system
+
+        cls.__system_types[expected_system] = cls
+
+    @model_validator(mode="wrap")
+    @classmethod
+    def validate_with_system(
+        cls, value: dict[str, Any], handler: ValidatorFunctionWrapHandler
+    ) -> Any:
+        """
+        Provides a model validator that instantiates the correct Identifier type based
+        on the supplied system. If either no system is provided, or the system provided
+        is not supported, the UnknownIdentifier type will be utilised.
+        """
+
+        if cls != Identifier or not isinstance(value, dict):
+            return handler(value)
+
+        system = value.get("system")
+        if system is None:
+            # This condition is unreachable as Pydantic will validate the presence of
+            # the required 'system' field before this validator is called.
+            raise ValueError("Identifier provided without a system attribute.")
+
+        identifier_cls = cls.__system_types.get(system)
+        if identifier_cls is None:
+            return UnknownIdentifier.model_validate(value)
+
+        return identifier_cls.model_validate(value)
+
+
+class UnknownIdentifier(Identifier, validate_system=False):
+    """Provides a fallback Identifier type for an unknown system."""
 
 
 class UUIDIdentifier(Identifier, expected_system="https://tools.ietf.org/html/rfc4122"):
     """A UUID identifier utilising the standard RFC 4122 system."""
 
-    def __init__(self, value: uuid.UUID | None = None):
-        super().__init__(
+    @classmethod
+    def create_with_uuid(cls, value: uuid.UUID | None = None) -> "UUIDIdentifier":
+        """Create a UUIDIdentifier with the provided UUID value."""
+        return cls(
             value=str(value or uuid.uuid4()),
-            system=self._expected_system,
+            system=cls.expected_system,
         )
 
 
@@ -79,15 +124,71 @@ class PatientIdentifier(
 ):
     """A FHIR R4 Patient Identifier utilising the NHS Number system."""
 
-    def __init__(self, value: str):
-        super().__init__(value=value, system=self._expected_system)
-
     @classmethod
     def from_nhs_number(cls, nhs_number: str) -> "PatientIdentifier":
         """Create a PatientIdentifier from an NHS number."""
-        return cls(value=nhs_number)
+        return cls(value=nhs_number, system=cls.expected_system)
+
+
+class OrganizationIdentifier(
+    Identifier, expected_system="https://fhir.nhs.uk/Id/ods-organization-code"
+):
+    """A FHIR R4 Organization Identifier utilising the ODS Organization Code system."""
+
+    @classmethod
+    def from_ods_code(cls, ods_code: str) -> "OrganizationIdentifier":
+        """Create an OrganizationIdentifier from an ODS code."""
+        return cls(value=ods_code, system=cls.expected_system)
 
 
 @dataclass(frozen=True)
 class LogicalReference[T: Identifier]:
     identifier: T
+
+
+@dataclass(frozen=True)
+class LiteralReference:
+    reference: str
+
+
+class Extension(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
+
+    __extension_types: ClassVar[dict[str, type["Extension"]]] = {}
+    type_name: ClassVar[str] = "__unknown__"
+
+    url: str = Field(..., frozen=True)
+
+    def __init_subclass__(cls, type_name: str) -> None:
+        cls.type_name = type_name
+        cls.__extension_types[type_name] = cls
+        super().__init_subclass__()
+
+    @model_validator(mode="wrap")
+    @classmethod
+    def validate_with_type(
+        cls, value: dict[str, Any], handler: ValidatorFunctionWrapHandler
+    ) -> Any:
+        """
+        Provides a model validator that instantiates the correct Extension type based on
+        the valueX field provided.
+        If an Extension subclass cannot be found, the default handler is utilised
+        instead.
+        """
+
+        # If we're not validating an Extension, or the value is not a dict,
+        # delegate to the default handler
+        if cls != Extension or not isinstance(value, dict):
+            return handler(value)
+
+        for key in value:
+            if key.startswith("value"):
+                type_name = key.split("value", 1)[1]
+                if (extension_cls := cls.__extension_types.get(type_name)) is not None:
+                    return extension_cls.model_validate(value)
+
+        return handler(value)
+
+
+class ReferenceExtension(Extension, type_name="Reference"):
+    value: LiteralReference = Field(..., alias="valueReference", frozen=True)