Merge remote-tracking branch 'origin/main' into feat/122-stream-endpoint-staging

# Conflicts:
#	src/api/app.py

Author: Shiven Pandya

README.md

@@ -177,3 +177,7 @@ Once the collections have been created, we search through each collection, linki
 # Start the FastAPI server
 uvicorn api.app:app --app-dir src --reload
 ```
+
+If you deploy in an environment where default temp directories are not writable
+(for example, some ECS task configurations), set `HSDS_TMP_DIR` to a writable
+path before starting the API.

src/api/app.py

@@ -20,6 +20,7 @@
     stage_multipart_uploads,
     validate_staged_workspace,
 )
+from api.tempdir import get_writable_temp_dir
 from lib.transform.collections import build_collections, searching_and_assigning
 from lib.transform.outputs import save_objects_to_json
 from api.model import HealthResponse
@@ -93,8 +94,13 @@ async def transform(
     except zipfile.BadZipFile:
         raise HTTPException(status_code=422, detail="Invalid zip file")
 
+    try:
+        temp_root = get_writable_temp_dir()
+    except RuntimeError as exc:
+        raise HTTPException(status_code=500, detail=str(exc))
+
     # Unzip into a temp directory
-    with tempfile.TemporaryDirectory() as input_dir:
+    with tempfile.TemporaryDirectory(dir=temp_root, prefix="hsds-input-") as input_dir:
         with zipfile.ZipFile(io.BytesIO(content), "r") as zf:
             zf.extractall(input_dir)
 
@@ -117,7 +123,7 @@ async def transform(
         results = searching_and_assigning(results)
 
         # Write each object to JSON files in another temp dir, then zip and return
-        with tempfile.TemporaryDirectory() as output_dir:
+        with tempfile.TemporaryDirectory(dir=temp_root, prefix="hsds-output-") as output_dir:
             save_objects_to_json(results, output_dir)
             buf = io.BytesIO()
             with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as out_zip:

src/api/tempdir.py

@@ -0,0 +1,42 @@
+import os
+import tempfile
+from pathlib import Path
+
+
+def _default_temp_candidates() -> list[Path]:
+    """Return ordered fallback temp-directory candidates."""
+    return [Path(tempfile.gettempdir()), Path("/tmp"), Path("/var/tmp"), Path("/usr/tmp")]
+
+
+def _is_writable_dir(path: Path) -> bool:
+    """Create directory (if needed) and verify write access."""
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except OSError:
+        return False
+
+    try:
+        with tempfile.NamedTemporaryFile(dir=path, prefix=".hsds-write-test-", delete=True):
+            pass
+    except OSError:
+        return False
+
+    return True
+
+
+def get_writable_temp_dir(env_var: str = "HSDS_TMP_DIR") -> str:
+    """Resolve a writable temp directory for API file operations."""
+    configured = os.getenv(env_var)
+    if configured:
+        candidate_paths = [Path(configured)]
+    else:
+        candidate_paths = _default_temp_candidates()
+
+    for candidate in candidate_paths:
+        if _is_writable_dir(candidate):
+            return str(candidate)
+
+    raise RuntimeError(
+        f"No writable temporary directory available. Set {env_var} to a writable path. "
+        f"Checked: {[str(path) for path in candidate_paths]}"
+    )

src/cli/main.py

@@ -1,19 +1,40 @@
+from pathlib import Path
+
 from ..lib.transform.outputs import save_objects_to_json
 from ..lib.transform.collections import build_collections, searching_and_assigning
 from ..lib.transform.logger import transformer_log
+from ..lib.custom_transform.transforms_loader import load_transforms_registry_if_available
 import click
 import sys
 
 @click.command()
 @click.argument('data_dictionary', type=click.Path(exists=True, file_okay=False, dir_okay=True, readable=True))
 @click.option('--output-dir', '-o', default='output', help='Output directory for JSON files')
 @click.option('--generate-ids', default=None, help='Generate new IDs using the provided organization name/id')
-
-def main(data_dictionary, output_dir, generate_ids):
+@click.option(
+    '--transforms',
+    type=click.Path(exists=False, dir_okay=False, file_okay=True, path_type=Path),
+    default=None,
+    help='Path to a Python module defining custom transforms (optional; omitted or missing file runs without them)',
+)
+
+def main(data_dictionary, output_dir, generate_ids, transforms):
     try:
         # Clear any previous log entries from prior runs
         transformer_log.clear()
 
+        transforms_registry = load_transforms_registry_if_available(transforms)
+        if transforms is None:
+            transformer_log.log("Custom transforms: not used (no --transforms path).")
+        elif transforms_registry is None:
+            transformer_log.log(
+                f"Custom transforms: not used (path not found or not a file: {transforms})."
+            )
+        else:
+            transformer_log.log(
+                f"Custom transforms: loaded from {transforms.resolve()}."
+            )
+
         results = build_collections(data_dictionary)  # Builds collections
         results = searching_and_assigning(results, requestor_identifier=generate_ids) # Links and cleans up, passes transformer_id
 

src/lib/custom_transform/__init__.py

@@ -0,0 +1,61 @@
+"""Custom exception for failures in user-provided transform/hook code."""
+
+from typing import Any
+
+
+class CustomTransformError(Exception):
+    """
+    Raised when an error occurs inside the user's custom transform or hook code.
+
+    Use this in a try/except around user code so callers can distinguish
+    user-script failures from transformer codebase failures. When re-raising,
+    pass the original exception as the cause to preserve the traceback::
+
+        try:
+            result = user_transform(value)
+        except Exception as e:
+            raise CustomTransformError(
+                "Field transformation failed",
+                function_name="clean_phone",
+                row_index=row_index,
+                cause=e,
+            ) from e
+
+    Context parameters (function_name, row_index, stage, etc.) are stored and
+    included in the exception message to help users debug their custom code.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        function_name: str | None = None,
+        row_index: int | None = None,
+        stage: str | None = None,
+        cause: BaseException | None = None,
+        **context: Any,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.function_name = function_name
+        self.row_index = row_index
+        self.stage = stage
+        self.cause = cause
+        self.context = context
+
+    def __str__(self) -> str:
+        parts = [
+            "Error in user-provided custom transform code.",
+            self.message,
+        ]
+        if self.function_name is not None:
+            parts.append(f"Function: {self.function_name!r}.")
+        if self.row_index is not None:
+            parts.append(f"Row index: {self.row_index}.")
+        if self.stage is not None:
+            parts.append(f"Stage: {self.stage!r}.")
+        for key, value in self.context.items():
+            parts.append(f"{key}: {value}.")
+        if self.cause is not None:
+            parts.append(f"Caused by: {type(self.cause).__name__}: {self.cause}.")
+        return " ".join(parts)

src/lib/custom_transform/custom_transform_error.py

@@ -0,0 +1,71 @@
+"""Handles loading user-provided custom modules. Its purpose is to dynamically
+import a single dedicated Python file defined by the user that contains their
+customized cleanup functions."""
+
+import importlib.util
+from pathlib import Path
+from typing import Callable
+
+__all__ = ["TransformsRegistry", "load_transforms_registry_if_available"]
+
+
+def load_transforms_registry_if_available(
+    module_path: str | Path | None,
+) -> "TransformsRegistry | None":
+    """
+    Load a TransformsRegistry from a file path, or return None when custom
+    transforms should not be used (no path, empty path, or missing file).
+    """
+    if module_path is None:
+        return None
+    if isinstance(module_path, str) and not module_path.strip():
+        return None
+    path = Path(module_path).expanduser()
+    try:
+        path = path.resolve()
+    except OSError:
+        return None
+    if not path.is_file():
+        return None
+    return TransformsRegistry(path)
+
+
+class TransformsRegistry:
+    """
+    Loads, stores, and accesses relevant information from a user-defined
+    custom-transforms module.
+    """
+
+    def __init__(self, module_path: Path):
+        """
+        Loads custom transforms module, loads transforms and hooks
+
+        Module must define:
+            transforms: dict
+            hooks     : dict
+
+        TODO: Implement error handling
+        """
+
+        module_name = module_path.stem
+        spec   = importlib.util.spec_from_file_location(module_name, module_path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+
+        self._transforms = module.transforms
+        self._hooks      = module.hooks
+
+
+    def get_transform(self, name: str) -> Callable:
+        """
+        Looks up specific field-level transformations by their string name.
+        TODO: Implement error handling
+        """
+        return self._transforms[name]
+
+    def get_hook(self, stage: str) -> Callable:
+        """
+        Retrieves broad row-level or collection-level hooks.
+        TODO: Implement error handling
+        """
+        return self._hooks[stage]

src/lib/custom_transform/transforms_loader.py

@@ -11,6 +11,7 @@
     "input_files_field",
     "split",
     "strip",
+    "transform",
     "description",
     "required",
 ]
@@ -26,4 +27,4 @@ def write_mapping_template_csv(rows: Iterable[FieldSpec], out_file: str) -> None
 
         for row in rows:
             required = "true" if row.required else "false"
-            writer.writerow([row.path, "", "", "", row.description, required])
+            writer.writerow([row.path, "", "", "", "", row.description, required])

src/lib/maintenance/generate_mapping/writer.py

@@ -6,7 +6,28 @@
 from pathlib import Path
 from typing import Callable
 
-__all__ = ["TransformsRegistry"]
+__all__ = ["TransformsRegistry", "load_transforms_registry_if_available"]
+
+
+def load_transforms_registry_if_available(
+    module_path: str | Path | None,
+) -> "TransformsRegistry | None":
+    """
+    Load a TransformsRegistry from a file path, or return None when custom
+    transforms should not be used (no path, empty path, or missing file).
+    """
+    if module_path is None:
+        return None
+    if isinstance(module_path, str) and not module_path.strip():
+        return None
+    path = Path(module_path).expanduser()
+    try:
+        path = path.resolve()
+    except OSError:
+        return None
+    if not path.is_file():
+        return None
+    return TransformsRegistry(path)
 
 
 class TransformsRegistry:

src/lib/transform/custom_transform/transforms_loader.py

@@ -1,13 +1,13 @@
-path,input_files_field,split,strip,description,required
-,,,,,
-id,,,,Identifier,true
-name,,,,Name,false
-details,,,,Details,true
-details.summary,,,,Summary,true
-details.notes,,,,Notes,false
-tags[],,,,Tags,true
-contacts[],,,,Contacts,false
-contacts[].email,,,,Email,false
-contacts[].phone,,,,Phone,false
-contacts[].addresses[],,,,Addresses,false
-contacts[].addresses[].line1,,,,Line 1,false
+path,input_files_field,split,strip,transform,description,required
+,,,,,,
+id,,,,,Identifier,true
+name,,,,,Name,false
+details,,,,,Details,true
+details.summary,,,,,Summary,true
+details.notes,,,,,Notes,false
+tags[],,,,,Tags,true
+contacts[],,,,,Contacts,false
+contacts[].email,,,,,Email,false
+contacts[].phone,,,,,Phone,false
+contacts[].addresses[],,,,,Addresses,false
+contacts[].addresses[].line1,,,,,Line 1,false