with prefect.Flow(
    name=flow_config.scheduler_flow_name,
    run_config=run_configs.UniversalRun(
        labels=flow_config.labels,
    ),
    terminal_state_handler=workflow_utils.workflow_terminal_state_handler,
    schedule=schedule,
) as scheduler_flow:

    workflow_id = register_workflow.register_workflow(
        flow_config.scheduler_flow_name,
    )

    # fetch all consumer configs for this provider/resource_type
    consumer_configs = extract_utils.fetch_consumer_configs_for_provider_resource(
        flow_config.provider_code,
        flow_config.resource_type,
    )

    # for each config, start a child flow that runs the ETL process
    # <https://docs-v1.prefect.io/core/idioms/flow-to-flow.html>
    mapped_flow_ids = create_flow_run.map(
        parameters=dict(
            consumer_code=consumer_configs,
            provider_code=prefect.unmapped(flow_config.provider_code),
            resource_type=prefect.unmapped(flow_config.resource_type),
            file_type=prefect.unmapped(flow_config.file_type),
            flow_labels=prefect.unmapped(flow_config.labels),
        ),
        flow_name=prefect.unmapped(
            EtlActionEnum.PIPELINE.value,
        ),
        project_name=prefect.unmapped(
            flow_config.global_config.project_name,
        ),
        labels=prefect.unmapped(flow_config.labels),
    )

    # wait for the mapped Flows to complete before exiting
    wait_for_flow_run.map(
        mapped_flow_ids,
        raise_final_state=prefect.unmapped(True),
    )