chg: accunulate multiple params to list and logs

Author: taheeraahmed

src/datasync/nva/nva_search_resources.py

@@ -61,8 +61,8 @@ def search_resources_api(
     """
     Get resources from NVA API
 
-    This will first fetch the data from the NVA API based on the search parameters
-    then write the data to the specified S3 location
+    This will use the NVA API to fetch resources based on the provided parameters.
+    The data will be written to the specified S3 location in parquet format.
 
     Args:
         resource_name: Will be used for the output files on S3, avoid using '-'.
@@ -91,7 +91,7 @@ def search_resources_api(
     # Filter by contributor and publication year:
     uv run datasync nva search-resources-api \
         --resource-name "author-publications" \
-        --filter contributor="https://api.nva.unit.no/cristin/person/1773250" \\
+        --filter contributor="https://api.nva.unit.no/cristin/person/1773250" \
         --filter publication_year_since=2020
     """
     log.debug("Starting NVA API search with parameters", filters=filters)
@@ -119,7 +119,14 @@ def search_resources_api(
             log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
             raise typer.Exit(code=1)
 
-        search_params[key] = value
+        if key in search_params:
+            existing = search_params[key]
+            if isinstance(existing, list):
+                existing.append(value)
+            else:
+                search_params[key] = [existing, value]
+        else:
+            search_params[key] = value
     log.debug("Parsed search parameters", search_params=search_params)
     if not search_params:
         log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
@@ -143,7 +150,7 @@ def search_resources_api(
     )
 
     log.info(f"Fetching resources from {base_url}")
-    load_info = pipeline.run(
+    pipeline.run(
         nva_search_source(
             base_url=base_url,
             resource_name=resource_name,
@@ -153,8 +160,6 @@ def search_resources_api(
         loader_file_format="parquet",
     )
 
-    log.info("Pipeline run completed", load_info=load_info)
-
     if apply_filter:
         con = setup_duckdb_s3_connection(
             endpoint_url=storage_endpoint_url,
@@ -174,7 +179,6 @@ def search_resources_api(
     write_timestamp(con, storage_bucket, storage_prefix)
     con.close()
 
-    log.info("NVA data sync completed")
     log.info(
         f"Data available at: {storage_endpoint_url}/{storage_bucket}/{storage_prefix}/"
         f"main/{resource_name}.parquet"

src/datasync/nva/utils.py

@@ -13,7 +13,7 @@
 
 def get_nva_resources(
     client: RESTClient,
-    search_params: dict[str, str],
+    search_params: dict[str, str | list[str]],
 ):
     """
     Function to fetch resources from NVA search API.
@@ -38,7 +38,7 @@ def get_nva_resources(
 def nva_search_source(
     base_url: str,
     resource_name: str = "resources",
-    search_params: dict[str, str] | None = None,
+    search_params: dict[str, str | list[str]] | None = None,
 ):
     """
     DLT source for fetching resources from NVA search API.
@@ -203,5 +203,7 @@ def apply_filter_transformation(
         compression="zstd",
         overwrite=True,
     )
-
-    log.info("Filter applied and data written back to S3")
+    log.info(
+        "Filter applied and data written back to S3",
+        number_of_resources=len(filtered_resources),
+    )