NRL-1798 Add dataclass and update tests

Author: sandyforresternhs

scripts/delete_pointers_by_id.py

@@ -2,6 +2,7 @@
 import json
 import os
 import tempfile
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, List
 
@@ -19,89 +20,109 @@ def _load_pointers_from_file(pointers_file: str) -> list[str]:
 
     Returns a list of pointer id strings. Prints a warning for skipped malformed JSON entries.
     """
-    with open(pointers_file, "r") as fh:
-        content = fh.read().strip()
+    with open(pointers_file, "r") as file:
+        content = file.read().strip()
 
     if not content:
         return []
 
-    # JSON path
     if content.startswith("[") or content.startswith("{"):
-        try:
-            data = json.loads(content)
-        except json.JSONDecodeError as e:
-            raise ValueError(f"Failed to parse JSON file {pointers_file}: {e}") from e
-
-        if not isinstance(data, list):
-            raise ValueError("JSON file must contain an array of objects")
-
-        parsed_ids: list[str] = []
-        skipped_count = 0
-        for item in data:
-            if (
-                isinstance(item, dict)
-                and "id" in item
-                and isinstance(item["id"], str)
-                and item["id"].strip()
-            ):
-                parsed_ids.append(item["id"].strip())
-            else:
-                skipped_count += 1
+        return _parse_json_pointers(content, pointers_file)
+
+    return _parse_plain_text_pointers(content)
 
-        if skipped_count:
-            print(
-                f"Warning: skipped {skipped_count} malformed entries in JSON file {pointers_file}"
-            )
 
-        return parsed_ids
+def _parse_json_pointers(content: str, pointers_file: str) -> list[str]:
+
+    try:
+        data = json.loads(content)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Failed to parse JSON file {pointers_file}: {e}") from e
+
+    if not isinstance(data, list):
+        raise ValueError("JSON file must contain an array of objects")
+
+    parsed_ids: list[str] = []
+    skipped_count = 0
+
+    for item in data:
+        if _is_valid_pointer(item):
+            parsed_ids.append(item["id"].strip())
+        else:
+            skipped_count += 1
 
-    # Plain text fallback
+    if skipped_count:
+        # TODO should we log these?
+        print(
+            f"Warning: skipped {skipped_count} malformed entries in JSON file {pointers_file}"
+        )
+
+    return parsed_ids
+
+
+def _is_valid_pointer(item: Any) -> bool:
+    return (
+        isinstance(item, dict)
+        and "id" in item
+        and isinstance(item["id"], str)
+        and item["id"].strip()
+    )
+
+
+def _parse_plain_text_pointers(content: str) -> list[str]:
     return [line.strip() for line in content.splitlines() if line.strip()]
 
 
-def _build_and_write_result(
-    pointers_to_delete,
-    ods_code,
-    matched_pointers,
-    mismatched_pointers,
-    not_found_pointers,
-    pointers_deleted,
-    failed_deletes,
-    start_time,
-    end_time,
-    output_file_path,
-):
+@dataclass
+class PointerDeletionContext:
+    pointers_to_delete: list[str]
+    ods_code: str
+    matched_pointers: list[str]
+    mismatched_pointers: list[str]
+    not_found_pointers: list[str]
+    pointers_deleted: list[str]
+    failed_deletes: list[str]
+    start_time: datetime
+    end_time: datetime
+    output_file_path: str
+
+
+def _build_and_write_result(ctx: PointerDeletionContext) -> Dict[str, Any]:
+
     result = {
-        "pointers_to_delete": len(pointers_to_delete),
-        "ods_code": ods_code,
-        "ods_code_matched": {"count": len(matched_pointers), "ids": matched_pointers},
+        "pointers_to_delete": len(ctx.pointers_to_delete),
+        "ods_code": ctx.ods_code,
+        "ods_code_matched": {
+            "count": len(ctx.matched_pointers),
+            "ids": ctx.matched_pointers,
+        },
         "ods_code_mismatched": {
-            "count": len(mismatched_pointers),
-            "ids": mismatched_pointers,
+            "count": len(ctx.mismatched_pointers),
+            "ids": ctx.mismatched_pointers,
         },
         "pointer_not_found": {
-            "count": len(not_found_pointers),
-            "ids": not_found_pointers,
+            "count": len(ctx.not_found_pointers),
+            "ids": ctx.not_found_pointers,
         },
-        "deleted_pointers": {"count": len(pointers_deleted), "ids": pointers_deleted},
-        "failed_deletes": {"count": len(failed_deletes), "ids": failed_deletes},
-        "deletes-took-secs": timedelta.total_seconds(end_time - start_time),
+        "deleted_pointers": {
+            "count": len(ctx.pointers_deleted),
+            "ids": ctx.pointers_deleted,
+        },
+        "failed_deletes": {"count": len(ctx.failed_deletes), "ids": ctx.failed_deletes},
+        "deletes-took-secs": timedelta.total_seconds(ctx.end_time - ctx.start_time),
     }
     try:
-        _write_result_file(result, output_file_path)
+        _write_result_file(result, ctx.output_file_path)
     except Exception as exc:
         result["_output_error"] = (
-            f"Failed to write result file {output_file_path}: {exc}"
+            f"Failed to write result file {ctx.output_file_path}: {exc}"
         )
     _print_summary(result)
     return result
 
 
 def _write_result_file(result: Dict[str, Any], output_file: str) -> None:
-    """
-    Atomically write result dict to output_file as JSON.
-    Raises on failure.
-    """
+
     out_dir = os.path.dirname(os.path.abspath(output_file)) or "."
     with tempfile.NamedTemporaryFile(
         "w", delete=False, dir=out_dir, prefix=".tmp_delete_results_", suffix=".json"
@@ -115,10 +136,7 @@ def _write_result_file(result: Dict[str, Any], output_file: str) -> None:
 def _check_pointers_match_ods_code(
     ods_code: str, pointer_ids: List[str]
 ) -> tuple[List[str], List[str]]:
-    """
-    Validate that pointer IDs are in line with the provided ODS code.
-    Returns (matched_ids, mismatched_ids)
-    """
+
     matched = []
     mismatched = []
 
@@ -173,7 +191,6 @@ def _batch_delete_pointers(
 ) -> tuple[List[str], List[str]]:
     """
     Delete pointers using BatchWriteItem (max 25 items per request).
-    Returns (deleted_ids, failed_ids)
     """
     pointers_deleted = []
     failed_deletes_set: set[str] = set()
@@ -212,9 +229,6 @@ def _batch_delete_pointers(
 
 
 def _print_summary(result: Dict[str, Any]) -> None:
-    """
-    Print a concise summary of the result to stdout.
-    """
 
     def count_from(field):
         val = result.get(field)
@@ -256,38 +270,46 @@ def _delete_pointers_by_id(
     - ods_code: ODS code of the organisation that the pointers belong to
     - pointers_to_delete: list of pointer ids to delete
     - pointers_file: path to JSON file (array of objects with "id" field) or text file (one id per line)
+
+    Sample usage:
+    - Delete by list of ids:
+      python delete_pointers_by_id.py --table_name MyTable --ods_code ABC123 --pointers_to_delete '["ABC123-12345678910", "ABC123-109876543210"]'
+    - Delete by JSON file:
+      python delete_pointers_by_id.py --table_name MyTable --ods_code ABC123 --pointers_file /path/to/pointers.json
+    - Delete by text file:
+      python delete_pointers_by_id.py --table_name MyTable --ods_code ABC123 --pointers_file /path/to/ids.txt
     """
     if pointers_to_delete is None and pointers_file is None:
         raise ValueError("Provide either pointers_to_delete or pointers_file")
 
     if pointers_to_delete is not None and pointers_file is not None:
         raise ValueError("Provide either pointers_to_delete or pointers_file, not both")
 
-    # Load pointers from file if provided
     if pointers_file:
         pointers_to_delete = _load_pointers_from_file(pointers_file)
 
-    # establish start_time early so any early-return can use it for filename
     start_time = datetime.now(tz=timezone.utc)
-    stamp = start_time.strftime("%Y%m%dT%H%M%SZ")
+    timestamp = start_time.strftime("%Y%m%dT%H%M%SZ")
     script_dir = os.path.dirname(os.path.abspath(__file__)) or "."
     output_file_path = os.path.join(
-        script_dir, f"delete_results_{ods_code}_{stamp}.json"
+        script_dir, f"delete_results_{ods_code}_{timestamp}.json"
     )
 
     if not pointers_to_delete:
         end_time = datetime.now(tz=timezone.utc)
         return _build_and_write_result(
-            pointers_to_delete,
-            ods_code,
-            [],
-            [],
-            [],
-            [],
-            [],
-            start_time,
-            end_time,
-            output_file_path,
+            PointerDeletionContext(
+                pointers_to_delete=pointers_to_delete,
+                ods_code=ods_code,
+                matched_pointers=[],
+                mismatched_pointers=[],
+                not_found_pointers=[],
+                pointers_deleted=[],
+                failed_deletes=[],
+                start_time=start_time,
+                end_time=end_time,
+                output_file_path=output_file_path,
+            )
         )
 
     print(
@@ -305,16 +327,18 @@ def _delete_pointers_by_id(
         print(f"None of the pointer IDs are a match for ODS code {ods_code}. Exiting.")
         end_time = datetime.now(tz=timezone.utc)
         return _build_and_write_result(
-            pointers_to_delete,
-            ods_code,
-            matched_pointers,
-            mismatched_pointers,
-            [],
-            [],
-            [],
-            start_time,
-            end_time,
-            output_file_path,
+            PointerDeletionContext(
+                pointers_to_delete=pointers_to_delete,
+                ods_code=ods_code,
+                matched_pointers=matched_pointers,
+                mismatched_pointers=mismatched_pointers,
+                not_found_pointers=[],
+                pointers_deleted=[],
+                failed_deletes=[],
+                start_time=start_time,
+                end_time=end_time,
+                output_file_path=output_file_path,
+            )
         )
 
     print(f"Checking existence of {len(matched_pointers)} pointers in {table_name}...")
@@ -330,16 +354,18 @@ def _delete_pointers_by_id(
         print("No pointers found to delete. Exiting.")
         end_time = datetime.now(tz=timezone.utc)
         return _build_and_write_result(
-            pointers_to_delete,
-            ods_code,
-            matched_pointers,
-            mismatched_pointers,
-            not_found_pointers,
-            [],
-            [],
-            start_time,
-            end_time,
-            output_file_path,
+            PointerDeletionContext(
+                pointers_to_delete=pointers_to_delete,
+                ods_code=ods_code,
+                matched_pointers=matched_pointers,
+                mismatched_pointers=mismatched_pointers,
+                not_found_pointers=not_found_pointers,
+                pointers_deleted=[],
+                failed_deletes=[],
+                start_time=start_time,
+                end_time=end_time,
+                output_file_path=output_file_path,
+            )
         )
 
     # Proceed with deletion using BatchWriteItem
@@ -349,16 +375,18 @@ def _delete_pointers_by_id(
 
     end_time = datetime.now(tz=timezone.utc)
     result = _build_and_write_result(
-        pointers_to_delete,
-        ods_code,
-        matched_pointers,
-        mismatched_pointers,
-        not_found_pointers,
-        pointers_deleted,
-        failed_deletes,
-        start_time,
-        end_time,
-        output_file_path,
+        PointerDeletionContext(
+            pointers_to_delete=pointers_to_delete,
+            ods_code=ods_code,
+            matched_pointers=matched_pointers,
+            mismatched_pointers=mismatched_pointers,
+            not_found_pointers=not_found_pointers,
+            pointers_deleted=pointers_deleted,
+            failed_deletes=failed_deletes,
+            start_time=start_time,
+            end_time=end_time,
+            output_file_path=output_file_path,
+        )
     )
     print(" Done")
     return result