diff --git a/src/fxhnt/adapters/orchestration/definitions.py b/src/fxhnt/adapters/orchestration/definitions.py new file mode 100644 index 0000000..da4fbaf --- /dev/null +++ b/src/fxhnt/adapters/orchestration/definitions.py @@ -0,0 +1,30 @@ +"""Dagster Definitions for the B0 lifecycle graph — the 4 combined-book assets + a daily schedule that +materializes the track (replaces the fxhnt-forward cron, same 23:30 UTC slot).""" +from __future__ import annotations + +from dagster import Definitions, ScheduleDefinition, define_asset_job + +from fxhnt.adapters.orchestration.assets import ( + cockpit_forward, + combined_forward_nav, + crypto_bars, + futures_bars, +) + +combined_book_job = define_asset_job( + name="combined_book_forward_job", + selection=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward], +) + +daily_combined_book_schedule = ScheduleDefinition( + name="combined_book_daily", + job=combined_book_job, + cron_schedule="30 23 * * *", + execution_timezone="UTC", +) + +defs = Definitions( + assets=[crypto_bars, futures_bars, combined_forward_nav, cockpit_forward], + jobs=[combined_book_job], + schedules=[daily_combined_book_schedule], +) diff --git a/tests/integration/test_orchestration_definitions.py b/tests/integration/test_orchestration_definitions.py new file mode 100644 index 0000000..5ac0d3b --- /dev/null +++ b/tests/integration/test_orchestration_definitions.py @@ -0,0 +1,20 @@ +"""The Definitions load and expose a daily schedule over the combined-book asset job + the 4 assets.""" +from __future__ import annotations + + +def test_definitions_load_with_assets_and_schedule() -> None: + from fxhnt.adapters.orchestration.definitions import defs + + # --- assets present --- + # defs.get_repository_def() is available in Dagster 1.12; its .assets_defs_by_key + # maps AssetKey → AssetsDefinition for every registered asset. + repo = defs.get_repository_def() + asset_keys = {k.to_user_string() for k in repo.assets_defs_by_key} + assert {"crypto_bars", "futures_bars", "combined_forward_nav", "cockpit_forward"} <= asset_keys + + # --- schedule present with the right cron --- + # defs.schedules is a list[ScheduleDefinition] (or None when empty) + scheds = list(defs.schedules or []) + assert any(getattr(s, "cron_schedule", "") == "30 23 * * *" for s in scheds), ( + f"expected a schedule with cron '30 23 * * *'; got {[getattr(s, 'cron_schedule', None) for s in scheds]}" + )