chg: accumulate multiple params to list and logs

Author: taheeraahmed

src/datasync/nva/nva_search_resources.py

@@ -1,3 +1,5 @@
+from collections import defaultdict
+
 import typer
 
 from ..settings import (
@@ -61,8 +63,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 +93,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)
@@ -109,7 +111,7 @@ def search_resources_api(
             "in the output file names due to DLT naming conventions."
         )
 
-    search_params = {}
+    search_params: defaultdict[str, list[str]] = defaultdict(list)
     for filter_str in filters:
         key, value = filter_str.split("=", 1)
         key = key.strip()
@@ -119,7 +121,8 @@ def search_resources_api(
             log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
             raise typer.Exit(code=1)
 
-        search_params[key] = value
+        search_params[key].append(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,18 +146,16 @@ 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,
-            search_params=search_params,
+            **search_params,
         ),
         write_disposition="replace",
         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 +175,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, list[str]],
 ):
     """
     Function to fetch resources from NVA search API.
@@ -38,42 +38,39 @@ def get_nva_resources(
 def nva_search_source(
     base_url: str,
     resource_name: str = "resources",
-    search_params: dict[str, str] | None = None,
+    **kwargs: list[str],
 ):
     """
     DLT source for fetching resources from NVA search API.
 
     Args:
         base_url: NVA API base URL
         resource_name: Name for the DLT resource
-        search_params: Dictionary of search parameters (project, publisher, etc.)
+        **kwargs: Search parameters as keyword arguments (project, publisher, etc.)
 
     Example:
-        # For project-based search:
-        nva_search_source(
-            resource_name="renew_hydro_resources",
-            search_params={"project": "https://api.nva.unit.no/cristin/project/2732649"}
-        )
-
-        # For publisher-based search:
-        nva_search_source(
-            resource_name="atlantic_salmon_resources",
-            search_params={"publisher": "Vitenskapelig råd for lakseforvaltning"}
-        )
-
-        # For more parameters: https://swagger-ui.nva.unit.no/
-    """
-    if search_params is None:
-        search_params = {}
+    # For project-based search:
+    nva_search_source(
+        resource_name="renew_hydro_resources",
+        project=["https://api.nva.unit.no/cristin/project/2732649"]
+    )
+
+    # For publisher-based search:
+    nva_search_source(
+        resource_name="atlantic_salmon_resources",
+        publisher=["Vitenskapelig råd for lakseforvaltning"]
+    )
 
+    # For more parameters: https://swagger-ui.nva.unit.no/
+    """
     client = RESTClient(
         base_url=base_url,
         paginator=JSONLinkPaginator(next_url_path="nextResults"),
         data_selector="hits",
     )
 
     yield dlt.resource(
-        get_nva_resources(client, search_params),
+        get_nva_resources(client, kwargs),
         name=resource_name,
         write_disposition="replace",
         primary_key="identifier",
@@ -203,5 +200,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),
+    )