from hyprster.dbt_jobs.resource import cfg
from dagster_dbt import load_assets_from_dbt_project
from dagster import ScheduleDefinition, AssetSelection,  define_asset_job, op


@op
def refresh_morning_external_tables(context):
    return context.resources.dbt.run_operation("stage-external-sources --exclude received_campaigns")

dbt_freshness_5am_assets = load_assets_from_dbt_project(
    project_dir = cfg['project_dir'],
    profiles_dir = cfg['profiles_dir'],
    select = "tag:freshness_5am",
    key_prefix = ["freshness_5am"]
)

dbt_freshness_5am_job = define_asset_job(name = "update_dbt_assets_at_5am", selection = AssetSelection.groups("freshness_5am"))

dbt_freshness_5am_job_schedule = ScheduleDefinition(
    name="dbt_freshness_5am_job_schedule",
    job=dbt_freshness_5am_job,
    cron_schedule="0 5 * * *"
)