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,
):
    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 , ) : 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
Execute pipeline using message queue as a transport