python_modules/dagster-graphql/dagster_graphql/implementation/pipeline_execution_manager.py
def execute_pipeline_through_queue(
    repository_info,
    pipeline_name,
    solid_subset,
    environment_dict,
    run_id,
    message_queue,
    reexecution_config,
    step_keys_to_execute,
):
    """
    Execute pipeline using message queue as a transport
    """

    message_queue.put(ProcessStartedSentinel(os.getpid()))

    run_config = RunConfig(
        run_id,
        event_callback=message_queue.put,
        executor_config=InProcessExecutorConfig(raise_on_error=False),
        reexecution_config=reexecution_config,
        step_keys_to_execute=step_keys_to_execute,
    )

    repository_container = RepositoryContainer(repository_info)
    if repository_container.repo_error:
        message_queue.put(
            MultiprocessingError(
                serializable_error_info_from_exc_info(repository_container.repo_error)
            )
        )
        return

    try:
        result = execute_pipeline(
            repository_container.repository.get_pipeline(pipeline_name).build_sub_pipeline(
                solid_subset
            ),
            environment_dict,
            run_config=run_config,
        )
        return result
    except:  # pylint: disable=W0702
        error_info = serializable_error_info_from_exc_info(sys.exc_info())
        message_queue.put(MultiprocessingError(error_info))
    finally:
        message_queue.put(MultiprocessingDone())
        message_queue.close()
def execute_pipeline_through_queue(
    repository_info,
    pipeline_name,
    solid_subset,
    environment_dict,
    run_id,
    message_queue,
    reexecution_config,
    step_keys_to_execute,
):
    """
    Execute pipeline using message queue as a transport
    """

    message_queue.put(ProcessStartedSentinel(os.getpid()))

    run_config = RunConfig(
        run_id,
        event_callback=message_queue.put,
        executor_config=InProcessExecutorConfig(raise_on_error=False),
        reexecution_config=reexecution_config,
        step_keys_to_execute=step_keys_to_execute,
    )

    repository_container = RepositoryContainer(repository_info)
    if repository_container.repo_error:
        message_queue.put(
            MultiprocessingError(
                serializable_error_info_from_exc_info(repository_container.repo_error)
            )
        )
        return

    try:
        result = execute_pipeline(
            repository_container.repository.get_pipeline(pipeline_name).build_sub_pipeline(
                solid_subset
            ),
            environment_dict,
            run_config=run_config,
        )
        return result
    except:  # pylint: disable=W0702
        error_info = serializable_error_info_from_exc_info(sys.exc_info())
        message_queue.put(MultiprocessingError(error_info))
    finally:
        message_queue.put(MultiprocessingDone())
        message_queue.close()
Execute pipeline using message queue as a transport