以编程方式提交dagster (没有dagit graphQL)

Question

有没有一种方法，我可以提交达格斯特运行编程，而不使用达吉特graphQL？

    dagster_instance = DagsterInstance.get()
    dagster_instance.submit_run(
        pipeline_run.run_id,
        workspace=graphene_info.context
    )

基本上，我想使用上面的代码来提交一个新的运行，但我无法弄清楚如何获得pipeline_run和工作区。另外，我不打算使用Dagit GraphQL API。

如果还有其他方法，请提出建议。

感谢你的时间和帮助。

Answer 1

Dagster Python还有一个实验性的GraphQL客户机，您可以使用它。

博士：https://docs.dagster.io/concepts/dagit/graphql-client

from dagster_graphql import DagsterGraphQLClient
from dagster_graphql import DagsterGraphQLClientError
from dagster import PipelineRunStatus

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

# submitting a job run
try:
    new_run_id: str = client.submit_job_execution(
        JOB_NAME,
        repository_location_name=REPO_LOCATION_NAME,
        repository_name=REPO_NAME,
        run_config={},
    )
    do_something_on_success(new_run_id)
except DagsterGraphQLClientError as exc:
    do_something_with_exc(exc)
    raise exc

# getting a job's run status
try:
    status: PipelineRunStatus = client.get_run_status(RUN_ID)
    if status == PipelineRunStatus.SUCCESS:
        do_something_on_success()
    else:
        do_something_else()
except DagsterGraphQLClientError as exc:
    do_something_with_exc(exc)
    raise exc