NRL-1215 Detect duplicate fields #801

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

mattdean3-nhs merged 27 commits into develop from feature/axkr1-NRL-1215-duplicate-fields

Jan 30, 2025

layer/nrlf/core/errors.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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]
@@ Expand Down @@

layer/nrlf/core/json_duplicate_checker.py

-Original file line number
+Diff line change
@@ -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

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand Down @@

layer/nrlf/core/request.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand Down Expand Up / @@ -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
@@ Expand All / @@ -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,
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

NRL-1215 Detect duplicate fields #801

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!