scaffold sensor triggered html download

apps/stocks/src/assets.py

@@ -0,0 +1,12 @@
+from functools import partial
+
+from config import APP
+
+import dagster as dg
+
+asset = partial(dg.asset, key_prefix=APP)
+
+
+@asset
+def raw_html() -> None:
+    print("todo")

apps/stocks/src/config.py

@@ -0,0 +1,4 @@
+import os
+from pathlib import Path
+
+APP = os.environ.get("APP", Path(__file__).parent.parent.name)

apps/stocks/src/definitions.py

@@ -1,6 +1,7 @@
 import os
 
 import assets
+import sensors
 from dagster_polars import PolarsParquetIOManager
 from icecream import install
 
@@ -24,4 +25,5 @@ definitions = dg.Definitions(
             base_dir=os.environ.get("STORAGE_DIR", "/storage") + f"/{APP}"
         ),
     },
+    sensors=[sensors.check_update],
 )

apps/stocks/src/jobs.py

@@ -0,0 +1,5 @@
+import assets
+
+import dagster as dg
+
+raw_html_job = dg.define_asset_job("deals_job", selection=[assets.raw_html.key])

apps/stocks/src/sensors.py

@@ -0,0 +1,15 @@
+from collections.abc import Iterator
+from datetime import date
+
+import jobs
+
+import dagster as dg
+
+
+@dg.sensor(job=jobs.raw_html_job)
+def check_update(context: dg.SensorEvaluationContext) -> Iterator[dg.RunRequest]:
+    ic(context.cursor)
+
+    yield dg.RunRequest()
+
+    context.update_cursor(date.today().strftime("%Y-%m-%d"))