feat: ⚡️ add job for dune upload

ggdp/__init__.py

@@ -1,11 +1,12 @@
 import os
 
-from dagster import Definitions, load_assets_from_modules
+from dagster import Definitions, load_assets_from_modules, define_asset_job
 from dagster_dbt import dbt_cli_resource, load_assets_from_dbt_project
 from dagster_duckdb import DuckDBResource
 from dagster_duckdb_pandas import DuckDBPandasIOManager
 
 from . import assets
+from . import ops
 
 DBT_PROJECT_DIR = os.path.dirname(os.path.abspath(__file__)) + "/../dbt/"
 
@@ -24,4 +25,17 @@
     ),
 }
 
-defs = Definitions(assets=[*dbt_assets, *all_assets], resources=resources)
+build_all_assets = define_asset_job(
+    name="build_all", selection="*", description="Materialize all assets"
+)
+build_indexer_assets = define_asset_job(
+    name="build_light_assets",
+    selection=["++rounds", "++projects"],
+    description="Materialize downstream of Gitcoin indexer-provided rounds, projects",
+)
+
+job_definitions = [ops.refresh_dune, build_all_assets, build_indexer_assets]
+
+defs = Definitions(
+    assets=[*dbt_assets, *all_assets], resources=resources, jobs=job_definitions
+)

ggdp/ops.py

@@ -0,0 +1,48 @@
+import io
+import json
+import os
+
+import pandas as pd
+import requests as rq
+
+from .assets import raw_allo_deployments
+from dagster import op, job, OpExecutionContext
+
+
+@op()
+def tiny_upload_to_dune(context: OpExecutionContext, data: pd.DataFrame) -> None:
+    """
+    Upload select dataframe to dune using API, needs DUNE_API_KEY envar to succeeed.
+    """
+
+    DUNE_KEY = os.getenv("DUNE_API_KEY")
+    if DUNE_KEY is None:
+        raise ValueError("DUNE_API_KEY envar not set, cannot upload")
+
+    file_buffer = io.StringIO()
+    data.to_csv(file_buffer, index=False)
+    file_buffer.seek(0)
+    csv_content = file_buffer.getvalue()
+
+    upload_payload = json.dumps(
+        {"table_name": "allo_contract_deployments", "data": csv_content}
+    )
+
+    response = rq.post(
+        "https://api.dune.com/api/v1/table/upload/csv",
+        headers={"X-Dune-Api-Key": DUNE_KEY},
+        data=upload_payload,
+    )
+    context.log.info(
+        f"new data uploaded to dune, result:{str(response.status_code)} | {str(response.text)}"
+    )
+
+    response.raise_for_status()
+
+
+@job
+def refresh_dune():
+    """
+    Refresh public dune dataset by materializing `raw_allo_deployments` and uploading new version.
+    """
+    tiny_upload_to_dune(data=raw_allo_deployments())