# Hourly from 8am Eastern through the end of the day, plus 4am
weekdays = CronClock("0 4,8-23 * * 1-5", start_date=pendulum.now("America/New_York"))
# every six hours on weekends
weekends = CronClock("0 */6 * * 0,6", start_date=pendulum.now("America/New_York"))
schedule = Schedule(clocks=[weekdays, weekends])

with Flow(
    "dbt Flow", schedule=schedule, state_handlers=[slack_failure_notification]
) as flow:
    select = Parameter("select", default=None)
    exclude = Parameter("exclude", default=None)
    full_refresh = Parameter("full_refresh", default=False)
    manifest = download_manifest(task_args={"name": "Download Manifest"})
    dbt_dependencies = execute_dbt(
        task_args={"name": "dbt Dependencies"}, command="dbt deps"
    )
    dbt_build = execute_dbt(
        task_args={"name": "dbt Build"},
        command=f"dbt build --exclude dbt_artifacts",
        select=select,
        exclude=exclude,
        full_refresh=full_refresh,
        upstream_tasks=[dbt_dependencies, manifest],
    )
    upload_manifest(task_args={"name": "Upload Manifest"}, upstream_tasks=[dbt_build])
    dbt_artifacts = execute_dbt(
        task_args={"name": "dbt Artifacts", "trigger": all_finished},
        command="dbt --no-write-json run-operation upload_dbt_artifacts_v2",
        upstream_tasks=[dbt_build],
    )
    execute_dbt(
        task_args={"name": "dbt Artifacts Build"},
        command="dbt build",
        select="dbt_artifacts",
        schema="dbt_artifacts_grebe",
        upstream_tasks=[dbt_artifacts],
    )
flow.set_reference_tasks([dbt_build])