Launch job/op with config run via graphql or python client example

[theelderbeever]

I found this way too difficult to get a straight answer on via the docs so I figured I would drop an example for future developers (and myself honestly). It is pieced together from various examples in the docs.

Tested against Dagster 1.6.8

Given a job/op definition as follows

from dagster import Config, OpExecutionContext, job, op


class DoSomethingConfig(Config):
    config_param: str


@op
def do_something(context: OpExecutionContext, config: DoSomethingConfig):
    context.log.info("config_param: " + config.config_param)


@job()
def do_it_all_with_default_config():
    do_something()

You can submit the job programmatically via

Python Client

The dagster python client using the following. You can optionally wrap the run_config in dagster.RunConfig

from dagster_graphql import DagsterGraphQLClient

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

client.submit_job_execution(
    "do_it_all_with_default_config",
    run_config={"ops": {"do_something": {"config": {"config_param": "things"}}}},
    # run_config=RunConfig(**{"ops": {"do_something": {"config": {"config_param": "things"}}}},)
)

GraphQL Mutation

# Write your query or mutation here
mutation LaunchRunMutation(
  $repositoryLocationName: String!
  $repositoryName: String!
  $jobName: String!
  $runConfigData: RunConfigData!
) {
  launchRun(
    executionParams: {
      selector: {
        repositoryLocationName: $repositoryLocationName
        repositoryName: $repositoryName
        jobName: $jobName
      }
      runConfigData: $runConfigData
    }
  ) {
    __typename
    ... on LaunchRunSuccess {
      run {
        runId
      }
    }
    ... on RunConfigValidationInvalid {
      errors {
        message
        reason
      }
    }
    ... on PythonError {
      message
    }
  }
}

with query variables

{
  "repositoryLocationName": "<Name of the code server that shows up on the left sidebar in the UI>",
  "repositoryName": "__repository__",
  "jobName": "do_it_all_with_default_config",
  "runConfigData": {
    "ops": {
      "do_something": {
        "config": {
          "config_param": "things"
        }
      }
    }
  }
}

[benpankow]

Submitting a Partitioned Run over GraphQL

As an addendum to @theelderbeever's answer, you can also submit partitioned runs over GraphQL using tags.

from dagster import Config, OpExecutionContext, job, op, DailyPartitionsDefinition, Definitions

daily_partition_def = DailyPartitionsDefinition(start_date="2024-01-01")


class DoSomethingConfig(Config):
    config_param: str


@op
def do_something(context: OpExecutionContext, config: DoSomethingConfig):
    context.log.info("config_param: " + config.config_param)


@job(partitions_def=daily_partition_def)
def do_it_all_with_default_config():
    do_something()

defs = Definitions(jobs=[do_it_all_with_default_config])

Python Client

from dagster_graphql import DagsterGraphQLClient

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

client.submit_job_execution(
    "do_it_all_with_default_config",
    run_config={"ops": {"do_something": {"config": {"config_param": "things"}}}},
    tags={"dagster/partition": "2024-01-05"}
)

GraphQL Mutation

mutation LaunchRunMutation(
  $repositoryLocationName: String!
  $repositoryName: String!
  $jobName: String!
  $runConfigData: RunConfigData!
  $tags:[ExecutionTag!]
) {
  launchRun(
    executionParams: {
      selector: {
        repositoryLocationName: $repositoryLocationName
        repositoryName: $repositoryName
        jobName: $jobName
      }
      runConfigData: $runConfigData
      executionMetadata: {
        tags: $tags
      }
    }
  ) {
    __typename
    ... on LaunchRunSuccess {
      run {
        runId
      }
    }
    ... on RunConfigValidationInvalid {
      errors {
        message
        reason
      }
    }
    ... on PythonError {
      message
    }
  }
}

with query variables

{
  "repositoryLocationName": "partition_demo.py",
  "repositoryName": "__repository__",
  "jobName": "do_it_all_with_default_config",
  "tags": {
    "key": "dagster/partition",
    "value": "2024-05-01"
  },
  "runConfigData": {
    "ops": {
      "do_something": {
        "config": {
          "config_param": "things"
        }
      }
    }
  }
}

[cloudymcrob]

this is very helpful as there is no explanation on how to do a partitioned run in the docs, thankyou!