chg(doffin): add doffin pipeline

Author: taheeraahmed

src/datasync/doffin.py

@@ -0,0 +1,101 @@
+import dlt
+import typer
+from dlt.destinations.impl.filesystem.factory import filesystem
+from dlt.sources.credentials import AwsCredentials
+from dlt.sources.rest_api import rest_api_source
+from dlt.sources.rest_api.config_setup import PageNumberPaginator
+
+from .settings import env, log
+
+app = typer.Typer(help="Export Doffin notices to Parquet in S3 bucket")
+
+DOFFIN_API_KEY = env("DOFFIN_API_KEY", default="")
+DOFFIN_BASE_URL = env("DOFFIN_BASE_URL", default="https://api.doffin.no/public/v2/")
+DOFFIN_PAGE_SIZE = int(env("DOFFIN_PAGE_SIZE", default="100"))
+
+DOFFIN_AWS_ACCESS_KEY = env("DOFFIN_AWS_ACCESS_KEY", default="")
+DOFFIN_AWS_SECRET_KEY = env("DOFFIN_AWS_SECRET_KEY", default="")
+DOFFIN_AWS_ENDPOINT = env("DOFFIN_AWS_ENDPOINT", default="")
+DOFFIN_AWS_BUCKET = env("DOFFIN_AWS_BUCKET", default="")
+DOFFIN_AWS_PREFIX = env("DOFFIN_AWS_PREFIX", default="doffin")
+DOFFIN_S3_REGION = env("DOFFIN_S3_REGION", default="us-east-1")
+
+
+@app.command()
+def run(
+    api_key: str = DOFFIN_API_KEY,
+    base_url: str = DOFFIN_BASE_URL,
+    page_size: int = DOFFIN_PAGE_SIZE,
+    access_key: str = DOFFIN_AWS_ACCESS_KEY,
+    secret_key: str = DOFFIN_AWS_SECRET_KEY,
+    endpoint_url: str = DOFFIN_AWS_ENDPOINT,
+    bucket: str = DOFFIN_AWS_BUCKET,
+    prefix: str = DOFFIN_AWS_PREFIX,
+    region: str = DOFFIN_S3_REGION,
+):
+    source = rest_api_source(
+        {
+            "client": {
+                "base_url": base_url,
+                "headers": {"Ocp-Apim-Subscription-Key": api_key},
+            },
+            "resources": [
+                {
+                    "name": "notices",
+                    "endpoint": {
+                        "path": "search",
+                        "params": {"numHitsPerPage": page_size},
+                        "data_selector": "hits",
+                        "paginator": PageNumberPaginator(
+                            page_param="page",
+                            base_page=1,
+                            total_path=None,
+                            # doffin api doesn't allow more than 1.000 rows
+                            # being fetched
+                            maximum_page=1000 // page_size,
+                        ),
+                    },
+                }
+            ],
+        }
+    )
+
+    credentials = AwsCredentials(
+        s3_url_style="path",
+        endpoint_url=endpoint_url,
+        aws_secret_access_key=secret_key,
+        aws_access_key_id=access_key,
+        region_name=region,
+    )
+
+    pipeline = dlt.pipeline(
+        pipeline_name="doffin",
+        destination=filesystem(
+            bucket_url=f"s3://{bucket}/{prefix}",
+            credentials=credentials,
+            layout="{table_name}.{ext}",
+        ),
+        dataset_name="doffin",
+    )
+
+    load_info = pipeline.run(
+        source, loader_file_format="parquet", write_disposition="replace"
+    )
+
+    row_counts = {}
+    if pipeline.last_trace and pipeline.last_trace.last_normalize_info:
+        row_counts = pipeline.last_trace.last_normalize_info.row_counts or {}
+
+    notices_rows_loaded = row_counts.get("notices", 0)
+
+    log.info(
+        "doffin load complete",
+        path=f"s3://{bucket}/{prefix}/notices.parquet",
+        load_id=load_info.loads_ids[0] if load_info.loads_ids else None,
+        rows_loaded=notices_rows_loaded,
+        row_counts=row_counts,
+    )
+
+
+if __name__ == "__main__":
+    app()

src/datasync/main.py

@@ -6,6 +6,7 @@
 
 from . import (
     dms,
+    doffin,
     gbif_backbone,
     grass,
     ninagen,
@@ -21,6 +22,7 @@
 )
 app.add_typer(nva.app, name="nva")
 app.add_typer(ubw.app, name="ubw")
+app.add_typer(doffin.app, name="doffin")
 app.add_typer(dms.app, name="dms")
 app.add_typer(ninagen.app, name="ninagen")
 app.add_typer(pit_registering_salmon.app, name="pit-registering-salmon")