How do I launch a pipeline from inside another pipeline?

[sryza]

How do I launch a pipeline from inside another pipeline?

[sryza]

⚠️ Note that this approach uses internal Dagster APIs and may break in a future release. ⚠️

We're planning to make this simpler and providing better compatibility guarantees.

from dagster_graphql.client.query import LAUNCH_PIPELINE_EXECUTION_MUTATION
from dagster_graphql.implementation.context import DagsterGraphQLContext
from dagster_graphql.test.utils import execute_dagster_graphql

from dagster import repository, solid
from dagster.cli.workspace.workspace import Workspace
from dagster.core.host_representation import RepositoryLocationHandle
from dagster.core.host_representation.selector import PipelineSelector


@repository
def my_repo():
    return [...]


@solid
def launch_downstream_pipeline(context):
    selector = PipelineSelector(
        location_name="<<in_process>>",
        repository_name=my_repo.name,
        pipeline_name="csv_hello_world",
        solid_selection=None,
    ).to_graphql_input()

    workspace = Workspace([RepositoryLocationHandle.create_in_process_location(my_repo.pointer)])
    run_config_data = {}

    graphql_context = DagsterGraphQLContext(workspace=workspace, instance=context.instance)
    execute_dagster_graphql(
        graphql_context,
        LAUNCH_PIPELINE_EXECUTION_MUTATION,
        variables={
            "executionParams": {
                "selector": selector,
                "runConfigData": run_config_data,
                "mode": "default",
                "executionMetadata": {
                    "tags": [{"key": "upstream_run_id", "value": context.run_id}],
                },
            }
        },
    )