NRL-1215 Detect duplicate fields

layer/nrlf/core/errors.py

@@ -79,6 +79,9 @@ def response(self) -> Response:
             body=self.operation_outcome.model_dump_json(exclude_none=True, indent=2),
         )
 
+    def __str__(self):
+        return f"OperationOutcomeError: {self.operation_outcome}"
+
 
 class ParseError(Exception):
     issues: List[OperationOutcomeIssue]

layer/nrlf/core/json_duplicate_checker.py

@@ -0,0 +1,74 @@
+import json
+from typing import Any
+
+
+def check_for_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    """Custom JSON object_pairs_hook that checks for duplicate keys."""
+    keys: dict[str, Any] = {}
+    dupes: dict[str, Any] = {}
+    normalized_keys: list[str] = []
+
+    for key, value in pairs:
+        normalized_key = key.lower()
+        if normalized_key in normalized_keys:
+            dupes.setdefault(key, []).append(value)
+        else:
+            keys[key] = value
+            normalized_keys += [normalized_key]
+
+    if dupes:
+        keys["__duplicates__"] = dupes
+
+    return keys
+
+
+def flatten_duplicates(data: dict[str, Any] | list[Any]) -> list[str]:
+    """Flattens a JSON structure and returns a list of duplicate paths."""
+    duplicates: list[str] = []
+    items = data.items() if isinstance(data, dict) else enumerate(data)
+
+    for key, value in items:
+        if key == "__duplicates__":
+            duplicates.extend(value.keys())
+        elif isinstance(value, (dict, list)):
+            path = f"{key}" if isinstance(data, dict) else f"[{key}]"
+            dupes = flatten_duplicates(value)
+            duplicates.extend([f"{path}.{dupe}" for dupe in dupes])
+
+    return duplicates
+
+
+def format_path(path: str) -> str:
+    """Transforms a path like key1.[2].key2 into key1[2].key2"""
+    parts = path.split(".")
+    formatted_parts: list[str] = []
+    for part in parts:
+        if part.startswith("["):
+            formatted_parts[-1] += part
+        else:
+            formatted_parts.append(part)
+    return ".".join(formatted_parts)
+
+
+def check_duplicate_keys(json_content: str) -> tuple[list[str], list[str]]:
+    """Find all duplicate keys in a JSON string.
+
+    Traverses the entire JSON structure and reports:
+    - List of keys that appear multiple times at the same level
+    - Full paths to each duplicate key occurrkeysence
+
+    A key is considered duplicate if it appears multiple times within
+    the same object, regardless of nesting level or array position.
+    """
+    try:
+        dupe_data = json.loads(json_content, object_pairs_hook=check_for_duplicate_keys)
+        duplicate_paths = [
+            f"DocumentReference.{format_path(path)}"
+            for path in flatten_duplicates(dupe_data)
+        ]
+        duplicate_keys = list(
+            dict.fromkeys([key.split(".")[-1] for key in duplicate_paths])
+        )
+        return duplicate_keys, duplicate_paths
+    except json.JSONDecodeError:
+        raise ValueError("Error: Invalid JSON format")

layer/nrlf/core/log_references.py

@@ -35,6 +35,8 @@ class LogReference(Enum):
     )
     HANDLER016 = _Reference("INFO", "Set response headers")
     HANDLER017 = _Reference("WARN", "Correlation ID not found in request headers")
+    HANDLER018 = _Reference("INFO", "Checking for duplicate keys in request body")
+    HANDLER019 = _Reference("ERROR", "Duplicate keys found in the request body")
     HANDLER999 = _Reference("INFO", "Request handler returned successfully")
 
     # Error Logs

layer/nrlf/core/request.py

@@ -6,6 +6,7 @@
 from nrlf.core.codes import SpineErrorConcept
 from nrlf.core.constants import CLIENT_RP_DETAILS, CONNECTION_METADATA
 from nrlf.core.errors import OperationOutcomeError, ParseError
+from nrlf.core.json_duplicate_checker import check_duplicate_keys
 from nrlf.core.logger import LogReference, logger
 from nrlf.core.model import ClientRpDetails, ConnectionMetadata
 
@@ -88,6 +89,7 @@ def parse_body(
 
     try:
         result = model.model_validate_json(body)
+        raise_when_duplicate_keys(body)
         logger.log(LogReference.HANDLER009, parsed_body=result.model_dump())
         return result
 
@@ -99,6 +101,24 @@ def parse_body(
         ) from None
 
 
+def raise_when_duplicate_keys(json_content: str) -> None:
+    """
+    Raises an error if duplicate keys are found in the JSON content.
+    """
+    logger.log(LogReference.HANDLER018)
+    duplicates, paths = check_duplicate_keys(json_content)
+    if duplicates:
+        error = OperationOutcomeError(
+            severity="error",
+            code="invalid",
+            details=SpineErrorConcept.from_code("MESSAGE_NOT_WELL_FORMED"),
+            diagnostics=f"Duplicate keys found in FHIR document: {duplicates}",
+            expression=paths,
+        )
+        logger.log(LogReference.HANDLER019, error=str(error))
+        raise error
+
+
 def parse_path(
     model: Type[BaseModel] | None,
     path_params: Dict[str, str] | None,