execute pipeline by api endpoint

[PlutarcoII]

Hi,

Is there a way to execute a pipeline using api endpoint?
Something like this:

from dagster import (
    pipeline, 
    solid, 
    execute_pipeline
)
from flask import Flask, request
import time
import threading

@solid
def hello_cereal(context):
    context.log.info(f' running from api endpoint')
    for _ in range(10):
        time.sleep(1)
        context.log.info(f'.')
    return "OK"

@pipeline
def hello_cereal_pipeline():
    hello_cereal()

app = Flask(__name__)

@app.route('/instance/apis/cereal', methods=['POST'])
def run_hello_cereal_pipeline():
    data = request.get_json()

    def run_pipeline(**kwargs):
        __params = kwargs.get('post_data', {})
        print("params:", __params)
        print("Starting pipeline")
        result = execute_pipeline(hello_cereal_pipeline)
        print(result) # OK

    thread = threading.Thread(target=run_pipeline, kwargs={'post_data': data})
    thread.start()
    return {"message": "Accepted"}, 202

Could you give me some advice?

thanks

[rexledesma]

There is a graphql interface over dagit where you can execute pipelines using a mutation. We also offer a client that abstracts out these graphql queries for you.

Checkout https://docs.dagster.io/concepts/dagit/graphql and https://docs.dagster.io/concepts/dagit/graphql-client for more details.

[PlutarcoII]

It works like a charm!

thanks

from dagster_graphql import DagsterGraphQLClient, DagsterGraphQLClientError
from dagster import PipelineRun
from datetime import datetime, time
from flask import Flask, request

client = DagsterGraphQLClient("localhost", port_number=3000)

app = Flask(__name__)
@app.route('/instance/apis/cereal', methods=['GET'])
def run_hello_cereal_pipeline():
    try:
        new_run_id: str = client.submit_pipeline_execution(
            'obten_acuses_coves_pipeline',
            repository_location_name='app.py',
            repository_name='vucem_repository',
            run_config= {
                "solids": {
                    "modelo": {
                        "config": {
                        "clientes": [],
                        "traficos": [''],
                        "start_date": datetime.now().strftime("%m/%d/%Y"),
                        "end_date": datetime.now().strftime("%m/%d/%Y"),
                        "action": 2 
                        }
                    }
                }
            },
            mode="default",
        )
    except DagsterGraphQLClientError as exc:
        print(exc)
        raise exc
    return {"message": new_run_id}, 202

if __name__ == "__main__":
    app.run(host="127.0.0.1", port="3001")

$ curl http://127.0.0.1:3001/instance/apis/cereal
returns:

{message: "6854542d-1799-4106-81a4-afb5bb30e28e"}