clean

Author: jonavellecuerdo

abdiff/cli.py

@@ -11,10 +11,10 @@
 from abdiff.config import Config, configure_logger
 from abdiff.core import (
     build_ab_images,
-    download_input_files,
     calc_ab_diffs,
     calc_ab_metrics,
     collate_ab_transforms,
+    download_input_files,
     init_run,
     run_ab_transforms,
 )
@@ -153,7 +153,7 @@ def init_job(
     "--download-files", is_flag=True, help="Pass to skip download of extract files"
 )
 def run_diff(
-    job_directory: str, input_files: str, message: str, download_files: bool
+    job_directory: str, input_files: str, message: str, *, download_files: bool
 ) -> None:
 
     job_data = read_job_json(job_directory)
@@ -174,7 +174,7 @@ def run_diff(
         image_tag_a=job_data["image_tag_a"],
         image_tag_b=job_data["image_tag_b"],
         input_files=input_files_list,
-        download_files=download_files,
+        use_local_s3=download_files,
     )
     collated_dataset_path = collate_ab_transforms(
         run_directory=run_directory,

abdiff/core/download_input_files.py

@@ -7,13 +7,32 @@
 
 from abdiff.config import Config
 
-
 logger = logging.getLogger(__name__)
 
 CONFIG = Config()
 
 
-def download_input_files(input_files: str):
+def download_input_files(input_files: list[str]) -> None:
+    """Download extract files from S3 to a local MinIO server.
+
+    For each file download, two AWS CLI commands are run by subprocess.
+    The output from the first command is piped to the second command.
+    These commands are further explained below:
+
+        1. Copy the contents from the input file and direct to stdout.
+           ```
+           aws s3 cp <input_file> -
+           ```
+
+        2. Given the stdout from the previous command as input, copy the contents
+           to a similarly named file on the local MinIO server.
+           ```
+           aws s3 cp --endpoint-url <minio_s3_url> --profile minio - <input_file>
+           ```
+
+    Note: An S3 client connected to the local MinIO server will check whether the file exists
+          prior to any download.
+    """
     s3_client = boto3.client(
         "s3",
         endpoint_url=CONFIG.minio_s3_url,
@@ -57,9 +76,10 @@ def check_object_exists(bucket: str, input_file: str, s3_client: S3Client) -> bo
     key = input_file.replace(f"s3://{bucket}/", "")
     try:
         s3_client.head_object(Bucket=bucket, Key=key)
-        return True
     except ClientError as exception:
         if exception.response["Error"]["Code"] == "404":
             return False
         logger.exception(f"Cannot determine if object exists for key {key}.")
         return False
+    else:
+        return True

abdiff/core/run_ab_transforms.py

@@ -34,8 +34,9 @@ def run_ab_transforms(
     image_tag_a: str,
     image_tag_b: str,
     input_files: list[str],
-    download_files: bool,
     docker_client: docker.client.DockerClient | None = None,
+    *,
+    use_local_s3: bool = False,
 ) -> tuple[list[str], ...]:
     """Run Docker containers with versioned images of Transmogrifier.
 
@@ -60,6 +61,10 @@ def run_ab_transforms(
             URIs for input files on S3 are accepted.
         docker_client (docker.client.DockerClient | None, optional): Docker client.
             Defaults to None.
+        use_local_s3 (bool): Boolean indicating whether the container should
+            access input files from a local MinIO server (i.e., "local S3 bucket")
+            or from AWS S3. This flag determines the appropriate environment variables
+            to set for the Docker containers. Default is False.
 
     Returns:
         tuple[list[str], ...]: A tuple containing two lists, where each list contains
@@ -97,7 +102,7 @@ def run_ab_transforms(
 
     # run containers and collect results
     futures = run_all_docker_containers(
-        docker_client, input_files, run_configs, download_files
+        docker_client, input_files, run_configs, use_local_s3=use_local_s3
     )
     containers, exceptions = collect_container_results(futures)
     logger.info(
@@ -132,7 +137,8 @@ def run_all_docker_containers(
     docker_client: docker.client.DockerClient,
     input_files: list[str],
     run_configs: list[tuple],
-    download_files: bool,
+    *,
+    use_local_s3: bool = False,
 ) -> list[Future]:
     """Invoke Docker containers to run in parallel via threads.
 
@@ -155,9 +161,12 @@ def run_all_docker_containers(
                     input_file,
                     get_transformed_filename(filename_details),
                     docker_client,
-                    download_files,
                 )
-                tasks.append(executor.submit(run_docker_container, *args))
+                tasks.append(
+                    executor.submit(
+                        run_docker_container, *args, use_local_s3=use_local_s3
+                    )
+                )
 
     logger.info(f"All {len(tasks)} containers have exited.")
     return tasks
@@ -170,16 +179,16 @@ def run_docker_container(
     input_file: str,
     output_file: str,
     docker_client: docker.client.DockerClient,
-    download_files: bool,
     timeout: int = CONFIG.transmogrifier_timeout,
+    *,
+    use_local_s3: bool = False,
 ) -> tuple[Container, Exception | None]:
     """Run Transmogrifier via Docker container to transform input file.
 
     The container is run in a detached state to capture a container handle for later use
     but this function waits for the container to exit before returning.
     """
-
-    if download_files:
+    if use_local_s3:
         environment_variables = {
             "AWS_ENDPOINT_URL": CONFIG.minio_s3_container_url,
             "AWS_ACCESS_KEY_ID": CONFIG.minio_root_user,