punchpipe.control.launcher#

Functions#

gather_planned_flows(session, weight_to_launch, ...)

count_flows(session, weights)

escalate_long_waiting_flows(session, pipeline_config)

determine_launchable_flow_count(weight_planned, ...)

launch_ready_flows(→ None)

Given a list of ready-to-launch flow_ids, this task creates flow runs in Prefect for them.

load_flow_data(pipeline_config)

launcher([pipeline_config_path])

The main launcher flow for Prefect, responsible for identifying flows, based on priority,

Module Contents#

punchpipe.control.launcher.gather_planned_flows(session, weight_to_launch, max_flows_to_launch, flow_weights, flow_enabled, flow_batch_sizes)[source]#
punchpipe.control.launcher.count_flows(session, weights)[source]#
punchpipe.control.launcher.escalate_long_waiting_flows(session, pipeline_config)[source]#
punchpipe.control.launcher.determine_launchable_flow_count(weight_planned, weight_running, max_weight_running, max_weight_to_launch, max_flows_to_launch)[source]#
async punchpipe.control.launcher.launch_ready_flows(session: sqlalchemy.orm.Session, flow_info: List[List[punchpipe.control.db.Flow]], tags_by_flow: dict[int, str], pipeline_config: dict) → None[source]#

Given a list of ready-to-launch flow_ids, this task creates flow runs in Prefect for them. These flow runs are automatically marked as scheduled in Prefect and will be picked up by a work queue and agent as soon as possible.

Parameters:

  • session (sqlalchemy.orm.session.Session) – A SQLAlchemy session for database interactions

  • flow_info (List[int]) – A list of flow IDs from the punchpipe database identifying which flows to launch

Return type:

A list of responses from Prefect about the flow runs that were created

punchpipe.control.launcher.load_flow_data(pipeline_config)[source]#
async punchpipe.control.launcher.launcher(pipeline_config_path=None)[source]#
The main launcher flow for Prefect, responsible for identifying flows, based on priority,

that are ready to run and creating flow runs for them. It also escalates long-waiting flows’ priorities.

See EM 41 or the internal requirements document for more details

Return type:

Nothing