The cloud-native orchestrator for the whole development lifecycle, with integrated lineage and observability.

dagster

Hi Team, when passing a run_request_for_partition are you able to include a run_config? When trying to pass it I get `TypeError: run_request_for_partition() got an unexpected keyword argument 'run_config'`

Hey team, we need the sensor to pass op configs basically

You can specify the op config on the job itself.

```def partition_fn(partition_key: str):
    return {"ops": {"my_op": {"config": {"partition": partition_key}}}} # here


@static_partitioned_config(partition_keys=["a", "b", "c", "d"])
def my_partitioned_config(partition_key: str):
    return partition_fn(partition_key)


@op
def my_op(context):
    print(context.op_config["partition"])


@job(config=my_partitioned_config)
def my_job():
    my_op()


@schedule(cron_schedule="* * * * *", job=my_job, execution_timezone="US/Pacific")
def my_schedule(_context):
    yield my_job.run_request_for_partition(partition_key="a", run_key=None)


@repository
def my_repo():
    return [my_job, my_schedule]```

<@UULA0R2LV> FYI I think they're using an asset job

we'd like to pass some vars somehow from the sensor to the job/op
```"ops": {
                                "unzip_and_write_to_gcs_date": {
                                    "config": {
                                        "zipfilepath": full_file_name ,
                                        "xmlfolder": run_key}}}```

this isn't currently possible, but I think it might be possible for us to add it in our next release. taking a look

would it be possible for your code to figure out full_file_name using the partition key? e.g. if you're launching a run for `2022-12-12`, the file name might be something like "zipfiles/20221212"

Oh wait, do we have access in ops to the run_key? We have it partitioned on date

Do we need to change the setup to be partitioned by file?

we just a different partitions definition?
```daily_funds_job = define_asset_job(
    "daily_funds_job",
    selection=AssetSelection.assets(fund_holdings_asset),
    partitions_def=daily_partitions_def)```


Oh I see, no we can't get the filename just from the date. We really need to pass it bc there are a few variations.

We can pass op configs if the job is not partitioned, though, right? Looks like RunRequest can get a `run_config`

Ah, got it. And yes, you can pass op configs if the job is not partitioned

Would working without partitions work for you in this case?

I think maybe a core issue here is that partitions in Dagster are tied closely with backfills - i.e. being able to go to an asset or job and kick off something that rematerializes all the partitions in history from scratch. But if the filenames aren't deterministic from the partition keys, then that backfill wouldn't be able to run

Yeah, I think we don't really need partitions