wip: nico helped with dlt

Author: taheeraahmed

src/datasync/nva.py

@@ -1,6 +1,11 @@
+import datetime
+import pathlib
+
 import dlt
 import typer
-from dlt.sources.rest_api import rest_api_source
+from dlt.destinations.impl.filesystem.factory import filesystem
+from dlt.sources.helpers.rest_client import RESTClient
+from dlt.sources.helpers.rest_client.paginators import JSONLinkPaginator
 
 from .settings import (
     INSTITUTION_CODE,
@@ -12,99 +17,42 @@
 app = typer.Typer()
 
 
-def dlt_source(
-    year,
-    resources,
-    projects,
-    persons,
-    categories,
-    funding_sources,
-    base_url,
-    institution_code,
-):
-    if year is None:
-        log.info("No year specified, syncing all available years.")
-        resource_params = {
-            "unit": institution_code,
-        }
-    else:
-        log.info(f"Syncing data for year: {year}")
-        resource_params = {
-            "unit": institution_code,
-            "publicationYearSince": year,
-            "publicationYearBefore": year + 1,
-        }
-    return rest_api_source(
-        {
-            "client": {
-                "base_url": base_url,
-                "paginator": {
-                    "type": "json_link",
-                    "next_url_path": "nextResults",
-                },
-            },
-            "resources": list(
-                filter(
-                    lambda r: r is not None,
-                    [
-                        {
-                            "name": "resources",
-                            "max_table_nesting": 0,
-                            "endpoint": {
-                                "path": "search/resources",
-                                "data_selector": "hits",
-                                "params": {
-                                    **resource_params,
-                                },
-                            },
-                        }
-                        if resources
-                        else None,
-                        {
-                            "name": "projects",
-                            "max_table_nesting": 0,
-                            "endpoint": {
-                                "path": f"cristin/organization/{institution_code}/projects",  # noqa: E501
-                                "data_selector": "hits",
-                            },
-                        }
-                        if projects
-                        else None,
-                        {
-                            "name": "persons",
-                            "max_table_nesting": 0,
-                            "endpoint": {
-                                "path": f"cristin/organization/{institution_code}/persons",  # noqa: E501
-                                "data_selector": "hits",
-                            },
-                        }
-                        if persons
-                        else None,
-                        {
-                            "name": "categories",
-                            "max_table_nesting": 0,
-                            "endpoint": {
-                                "path": "cristin/category/project",
-                                "data_selector": "hits",
-                            },
-                        }
-                        if categories
-                        else None,
-                        {
-                            "name": "funding_sources",
-                            "max_table_nesting": 0,
-                            "endpoint": {
-                                "path": "cristin/funding-sources",
-                                "data_selector": "hits",
-                            },
-                        }
-                        if funding_sources
-                        else None,
-                    ],
-                )
-            ),
-        }
-    )
+# {
+#     "name": "projects",
+#     "endpoint": {
+#         "path": f"cristin/organization/{institution_code}/projects",  # noqa: E501
+#         "data_selector": "hits",
+#     },
+# }
+# if projects
+# else None,
+# {
+#     "name": "persons",
+#     "endpoint": {
+#         "path": f"cristin/organization/{institution_code}/persons",  # noqa: E501
+#         "data_selector": "hits",
+#     },
+# }
+# if persons
+# else None,
+# {
+#     "name": "categories",
+#     "endpoint": {
+#         "path": "cristin/category/project",
+#         "data_selector": "hits",
+#     },
+# }
+# if categories
+# else None,
+# {
+#     "name": "funding_sources",
+#     "endpoint": {
+#         "path": "cristin/funding-sources",
+#         "data_selector": "hits",
+#     },
+# }
+# if funding_sources
+# else None,
 
 
 @app.command()
@@ -117,25 +65,49 @@ def run(
     base_url: str = NVA_BASE_URL,
     duckdb_name: str = NVA_DUCKDB_NAME,
     institution_code: str = INSTITUTION_CODE,
-    year: int = None,
 ):
+    @dlt.source()
+    def nva():
+        client = RESTClient(
+            base_url=base_url,
+            paginator=JSONLinkPaginator(next_url_path="nextResults"),
+            data_selector="hits",
+        )
+
+        def get_resources():
+            for year in set(range(2025, datetime.datetime.now().year + 1)):
+                log.debug("Fetching resources for year", year=year)
+                yield from client.paginate(
+                    "search/resources",
+                    method="get",
+                    params={
+                        "unit": institution_code,
+                        "publicationYearSince": year,
+                        "publicationYearBefore": year + 1,
+                    },
+                )
+
+        yield dlt.resource(
+            get_resources(),
+            name="resources",
+            write_disposition="replace",
+            max_table_nesting=1,
+        )
+
     pipeline = dlt.pipeline(
         pipeline_name=duckdb_name,
-        destination="duckdb",
+        destination=filesystem(
+            bucket_url="file://"
+            + str(pathlib.Path(__name__).absolute().parent / "data"),
+            layout="{table_name}.{ext}",
+        ),
         dataset_name="main",
+        progress="log",
     )
 
-    source = dlt_source(
-        year=year,
-        resources=resources,
-        projects=projects,
-        persons=persons,
-        categories=categories,
-        funding_sources=funding_sources,
-        base_url=base_url,
-        institution_code=institution_code,
+    log.info(
+        pipeline.run(nva(), write_disposition="replace", loader_file_format="parquet")
     )
-    log.info(pipeline.run(source))
 
 
 if __name__ == "__main__":