chg: make nva sync generic

Author: taheeraahmed

.env.example

@@ -5,20 +5,12 @@ DEBUG=false
 NVA_BASE_URL=https://api.nva.unit.no/
 NVA_DUCKDB_FILE_NAME=nva_sync
 NVA_INSTITUTION_CODE=7511.0.0.0
-NVA_ACCESS_KEY=
-NVA_SECRET_KEY=
-NVA_ENDPOINT=
-NVA_BUCKET=
-NVA_PREFIX=nva
-NVA_REGION=us-east-1
+NVA_S3_ACCESS_KEY=
+NVA_S3_SECRET_KEY=
+NVA_S3_ENDPOINT_URL=
+NVA_S3_BUCKET=
+NVA_S3_PREFIX=
 NVA_RESOURCES_DATA_S3_PATH=
-NVA_FILTER_STORAGE_ENDPOINT_URL=
-NVA_FILTER_STORAGE_ACCESS_KEY=
-NVA_FILTER_STORAGE_SECRET_KEY=
-NVA_FILTER_STORAGE_BUCKET=
-NVA_FILTER_STORAGE_PREFIX=nva-filtered
-NVA_FILTER_STORAGE_REGION=us-east-1
-NVA_FILTER_PATH_PARQUET_OUTPUT=
 
 # IPT (Integrated Publishing Toolkit)
 IPT_URL=https://ipt.nina.no

README.md

@@ -47,7 +47,7 @@ uvx --from . datasync
 ```
 
 ### Development
-Just run `uv run main.py` and you are good to go!
+Just run `uv run datasync --help` and see what you can do.
 
 ### Update from template
 To update your project with the latest changes from the template, run:

src/datasync/nva/__init__.py

@@ -2,12 +2,16 @@
 # this allows to attach the scripts to the app
 import typer
 
-from . import nva, nva_filtered
+from . import nva, nva_filter_resources, nva_search_resources
 
 app = typer.Typer(help="Commands to handle NVA tasks")
 
 app.command(help="Sync NVA data from REST API to target")(nva.run)
 
-app.command(help="Filter and create the tables from NVA data to parquet")(
-    nva_filtered.filter_data
+app.command(help="Filter NVA parquet files and export to S3")(
+    nva_filter_resources.filter_data
+)
+
+app.command(help="Fetch and filter NVA data with flexible search parameters")(
+    nva_search_resources.search_resources_api
 )

src/datasync/nva/nva.py

@@ -1,29 +1,28 @@
 import datetime
 
 import dlt
-import duckdb
 import typer
-from dlt.destinations.impl.filesystem.factory import filesystem
-from dlt.sources.credentials import AwsCredentials
 from dlt.sources.helpers.rest_client import RESTClient
 from dlt.sources.helpers.rest_client.paginators import JSONLinkPaginator
 
-from ..settings import (
-    env,
-    log,
+from ..settings import env, log
+from .settings import (
+    NVA_BASE_URL,
+    NVA_INSTITUTION_CODE,
+    NVA_S3_ACCESS_KEY,
+    NVA_S3_BUCKET,
+    NVA_S3_ENDPOINT_URL,
+    NVA_S3_PREFIX,
+    NVA_S3_SECRET_KEY,
+)
+from .utils import (
+    create_pipeline,
+    create_s3_credentials,
+    setup_duckdb_s3_connection,
+    write_timestamp,
 )
 
-NVA_BASE_URL = env("NVA_BASE_URL", default="https://api.nva.unit.no/")
 NVA_DUCKDB_NAME = env("NVA_DUCKDB_FILE_NAME", default="nva_sync")
-NVA_INSTITUTION_CODE = env("NVA_INSTITUTION_CODE", default="7511.0.0.0")
-
-NVA_ACCESS_KEY = env("NVA_ACCESS_KEY", default="")
-NVA_SECRET_KEY = env("NVA_SECRET_KEY", default="")
-NVA_ENDPOINT = env("NVA_ENDPOINT", default="")
-NVA_BUCKET = env("NVA_BUCKET", default="")
-
-NVA_PREFIX = env("NVA_PREFIX", default="nva")
-NVA_REGION = env("NVA_REGION", default="us-east-1")
 
 app = typer.Typer(help="Export NVA APIs to Parquet on a S3 Bucket")
 
@@ -95,7 +94,7 @@ def nva(
             get_resources(client, institution_code),
             name="resources",
             write_disposition="replace",
-            primary_key="identifier",
+            primary_key="id",
             max_table_nesting=1,
         )
 
@@ -104,6 +103,7 @@ def nva(
             get_projects(client, institution_code),
             name="projects",
             write_disposition="replace",
+            primary_key="id",
             max_table_nesting=1,
         )
 
@@ -112,20 +112,23 @@ def nva(
             get_persons(client, institution_code),
             name="persons",
             write_disposition="replace",
+            primary_key="id",
             max_table_nesting=1,
         )
     if categories:
         yield dlt.resource(
             get_categories(client),
             name="categories",
             write_disposition="replace",
+            primary_key="_dlt_id",
             max_table_nesting=1,
         )
 
     if funding_sources:
         yield dlt.resource(
             get_funding_sources(client),
             name="funding_sources",
+            primary_key="identifier",
             write_disposition="replace",
             max_table_nesting=1,
         )
@@ -143,12 +146,12 @@ def run(
     base_url: str = NVA_BASE_URL,
     duckdb_name: str = NVA_DUCKDB_NAME,
     institution_code: str = NVA_INSTITUTION_CODE,
-    endpoint_url: str = NVA_ENDPOINT,
-    access_key: str = NVA_ACCESS_KEY,
-    secret_key: str = NVA_SECRET_KEY,
-    bucket: str = NVA_BUCKET,
-    prefix: str = NVA_PREFIX,
-    region: str = NVA_REGION,
+    endpoint_url: str = NVA_S3_ENDPOINT_URL,
+    access_key: str = NVA_S3_ACCESS_KEY,
+    secret_key: str = NVA_S3_SECRET_KEY,
+    bucket: str = NVA_S3_BUCKET,
+    prefix: str = NVA_S3_PREFIX,
+    region: str = "us-east-1",
 ):
     if not endpoint_url:
         log.error("AWS S3 endpoint URL is not provided")
@@ -162,24 +165,20 @@ def run(
 
     log.info("Starting NVA data sync")
     log.info(f"Data will be available at: {endpoint_url}/{bucket}/{prefix}")
-    credentials = AwsCredentials(
-        s3_url_style="path",
+
+    credentials = create_s3_credentials(
         endpoint_url=endpoint_url,
-        aws_secret_access_key=secret_key,
-        aws_access_key_id=access_key,
-        region_name=region,
+        access_key=access_key,
+        secret_key=secret_key,
+        region=region,
     )
 
-    pipeline = dlt.pipeline(
+    pipeline = create_pipeline(
         pipeline_name=duckdb_name,
-        destination=filesystem(
-            region_name=region,
-            bucket_url=f"s3://{bucket}/" + prefix,
-            credentials=credentials,
-            layout="{table_name}.{ext}",
-        ),
-        dataset_name="main",
-        progress="log",
+        bucket=bucket,
+        prefix=prefix,
+        credentials=credentials,
+        region=region,
     )
 
     log.info(
@@ -201,28 +200,15 @@ def run(
     log.info("NVA data sync completed")
     log.info(f"Data available at: {endpoint_url}/{bucket}/{prefix}")
 
-    con = duckdb.connect()
-    con.execute("INSTALL httpfs;")
-    con.execute("LOAD httpfs;")
-    con.execute(f"""
-        CREATE OR REPLACE SECRET (
-            TYPE S3,
-            KEY_ID '{access_key}',
-            SECRET '{secret_key}',
-            ENDPOINT '{endpoint_url.replace("https://", "").replace("http://", "")}',
-            REGION '{region}',
-            URL_STYLE 'path'
-        );
-    """)
-
-    timestamp = datetime.datetime.now().isoformat()
-    con.execute(f"""
-        COPY (SELECT '{timestamp}' as last_successful_run)
-        TO 's3://{bucket}/{prefix}/last_successful_run.parquet'
-        (FORMAT PARQUET, COMPRESSION ZSTD)
-    """)
+    con = setup_duckdb_s3_connection(
+        endpoint_url=endpoint_url,
+        access_key=access_key,
+        secret_key=secret_key,
+        region=region,
+    )
+
+    write_timestamp(con, bucket, prefix)
     con.close()
-    log.info(f"Last successful run timestamp written: {timestamp}")
 
 
 if __name__ == "__main__":