scaffold nearby charging sites logic

apps/tesla/src/assets.py

@@ -0,0 +1,11 @@
+from functools import partial
+
+from shared.config import APP
+
+import dagster as dg
+
+asset = partial(dg.asset, key_prefix=APP)
+
+
+@asset
+def raw_nearby_charging_sites() -> None: ...

apps/tesla/src/definitions.py

@@ -1,4 +1,5 @@
 import assets
+import schedules
 from dagster_polars import PolarsParquetIOManager
 from icecream import install
 from shared.config import APP, STORAGE_DIR
@@ -19,4 +20,5 @@ definitions = dg.Definitions(
     resources={
         "polars_parquet_io_manager": PolarsParquetIOManager(base_dir=STORAGE_DIR),
     },
+    schedules=[schedules.raw_schedule],
 )

apps/tesla/src/jobs.py

@@ -0,0 +1,9 @@
+import assets
+
+import dagster as dg
+
+raw_nearby_charging_sites_job = dg.define_asset_job(
+    "raw_nearby_charging_sites_job",
+    selection=[assets.raw_nearby_charging_sites.key],
+    # partitions_def=assets.raw_nearby_charging_sites.partitions_def,
+)

apps/tesla/src/schedules.py

@@ -0,0 +1,15 @@
+import assets
+
+import dagster as dg
+
+raw_schedule = dg.ScheduleDefinition(
+    name="daily_refresh",
+    cron_schedule="*/10 * * * *",
+    target=[assets.raw_nearby_charging_sites],
+)
+
+# dg.build_schedule_from_partitioned_job(
+#     job=raw_nearby_charging_sites_job,
+#     hour_of_day=9,
+#     default_status=dg.DefaultScheduleStatus.STOPPED,
+# )