+generate DynamoDB data generate_dynamo_data.py

Author: ivma1-nhs

tests/e2e/data/generate_dynamo_data.py

@@ -2,38 +2,35 @@
 import logging
 import os
 import re
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 
-# Constants
 OUTPUT_ROOT = "out"
 DATE_FORMAT = "%Y%m%d"
 VAR_PATTERN = re.compile(r"<<([^<>]+)>>")
+REQUIRED_TOKEN_PARTS = 3
 
-# Configure logging
+logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
 
 
 class DateVariableResolver:
-    """Handles the logic for parsing and evaluating date-based variables."""
-
-    def __init__(self, today: datetime = None):
-        self.today = today or datetime.today()
+    def __init__(self, today: datetime | None = None):
+        self.today = today or datetime.now(tz=timezone.UTC)
 
     def resolve(self, token: str) -> str:
-        logging.debug(f"Resolving variable: {token}")
+        logger.debug("Resolving variable: %s", token)
         parts = token.split("_")
-        if len(parts) < 3 or parts[0].upper() != "DATE":
-            raise ValueError(f"Unsupported variable format: {token}")
-
+        if len(parts) < REQUIRED_TOKEN_PARTS or parts[0].upper() != "DATE":
+            msg = f"Unsupported variable format: {token}"
+            raise ValueError(msg)
         _, unit, value = parts[0], parts[1].lower(), parts[2]
-
         try:
             offset = int(value)
-        except ValueError:
-            raise ValueError(f"Invalid offset value: {value}")
-
+        except ValueError as err:
+            msg = f"Invalid offset value: {value}"
+            raise ValueError(msg) from err
         if unit == "day":
             return (self.today + timedelta(days=offset)).strftime(DATE_FORMAT)
         if unit == "week":
@@ -44,15 +41,12 @@ def resolve(self, token: str) -> str:
             try:
                 birth_date = self.today.replace(year=self.today.year - offset)
             except ValueError:
-                # Handle February 29th
                 birth_date = self.today.replace(month=2, day=28, year=self.today.year - offset)
             return birth_date.strftime(DATE_FORMAT)
-        raise ValueError(f"Unsupported calculation unit: {unit}")
-
+        msg = f"Unsupported calculation unit: {unit}"
+        raise ValueError(msg)
 
 class JsonTestDataProcessor:
-    """Processes JSON test files by resolving placeholders in 'data' arrays."""
-
     def __init__(self, input_dir: Path, output_dir: Path, resolver: DateVariableResolver):
         self.input_dir = input_dir
         self.output_dir = output_dir
@@ -71,56 +65,49 @@ def _replace_token(self, match: re.Match) -> str:
         token = match.group(1)
         try:
             return self.resolver.resolve(token)
-        except Exception as e:
-            logging.warning(f"Failed to resolve variable {token}: {e}")
+        except ValueError:
+            logger.warning("Failed to resolve variable: %s", token)
             return match.group(0)
 
     def process_file(self, file_path: Path):
-        logging.info(f"Processing file: {file_path}")
+        logger.info("Processing file: %s", file_path)
         try:
-            with open(file_path) as f:
+            with file_path.open() as f:
                 content = json.load(f)
-        except Exception as e:
-            logging.exception(f"Failed to read {file_path}: {e}")
+        except Exception:
+            logger.exception("Failed to read file: %s", file_path)
             return
-
         try:
             resolved = self.resolve_placeholders(content)
-        except Exception as e:
-            logging.exception(f"Failed to resolve placeholders: {e}")
+        except Exception:
+            logger.exception("Failed to resolve placeholders in file: %s", file_path)
             return
-
         if "data" not in resolved:
-            logging.error(f"Missing 'data' key in {file_path}")
+            logger.error("Missing 'data' key in file: %s", file_path)
             return
-
         relative_path = file_path.relative_to(self.input_dir)
         output_path = self.output_dir / relative_path
         output_path.parent.mkdir(parents=True, exist_ok=True)
-
         try:
-            with open(output_path, "w") as f:
+            with output_path.open("w") as f:
                 json.dump(resolved["data"], f, indent=2)
-            logging.info(f"Written resolved file: {output_path}")
-        except Exception as e:
-            logging.exception(f"Failed to write output: {e}")
-
+            logger.info("Written resolved file: %s", output_path)
+        except Exception:
+            logger.exception("Failed to write output to: %s", output_path)
 
 def main():
     input_dir = Path()
     output_dir = Path(OUTPUT_ROOT)
     resolver = DateVariableResolver()
-
     processor = JsonTestDataProcessor(input_dir, output_dir, resolver)
-
-    logging.info(f"Scanning for JSON files in {input_dir}")
+    logger.info("Scanning for JSON files in directory: %s", input_dir)
     for root, _, files in os.walk(input_dir):
         for file in files:
+            file_path = Path(root) / file
             if file.endswith(".json"):
-                processor.process_file(Path(root) / file)
+                processor.process_file(file_path)
             else:
-                logging.debug(f"Skipping non-JSON file: {file}")
-
+                logger.debug("Skipping non-JSON file: %s", file)
 
 if __name__ == "__main__":
     main()