feat(orchestration): Definitions + daily combined-book schedule
Wire the 4 B0 assets into combined_book_forward_job and a ScheduleDefinition (cron 30 23 * * *, UTC) via a Dagster Definitions object; dagster definitions validate passes. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
30
src/fxhnt/adapters/orchestration/definitions.py
Normal file
30
src/fxhnt/adapters/orchestration/definitions.py
Normal file
@@ -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],
|
||||
)
|
||||
20
tests/integration/test_orchestration_definitions.py
Normal file
20
tests/integration/test_orchestration_definitions.py
Normal file
@@ -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]}"
|
||||
)
|
||||
Reference in New Issue
Block a user