getting missing config errors when using partitions with dynamic graphs/jobs

[dagsir[bot]]

I am trying to use partitions with a dynamic graph/job, the dynamic job works fine but when I introduce the partition config as described in the the docs my dag starts throwing a config not provided error for the values for each of my ops, are there any docs/examples of how to make this combination work?

---

The question was originally asked in Dagster Slack.

[sryza]

Are you able to provide a minimal code example we can run that reproduces the problem?

[mathew-mwi]

Here is a sample code set that reproduces the issue:

from dagster import (
    op,
    job,
    DynamicOut,
    DynamicOutput,
    OpExecutionContext,
    daily_partitioned_config
)


from datetime import datetime

@op(out=DynamicOut())
def return_clients():
    clients = ["a", "b", "c"]
    for client in clients:
        yield DynamicOutput(
            value=client, mapping_key=client
        )

@op()
def echo(context: OpExecutionContext, client_id):
    context.log.info(f"echo {client_id} + {context.op_config.date}")
    return client_id

@daily_partitioned_config(start_date=datetime(2023, 11, 27))
def my_partitioned_config(start: datetime, _end: datetime):
    return {
        "ops": {
            "process_data_for_date": {"config": {"date": start.strftime("%Y-%m-%d")}}
        }
    }

@job(config=my_partitioned_config
        )
def demo():
    (
        return_clients()
        .map(echo)
        .collect()
    )

The error message is:

dagster._core.errors.DagsterInvalidConfigError: Error in config for job demo
Error 1: Received unexpected config entry "process_data_for_date" at path root:ops. Expected: "{ echo?: { config?: Any } return_clients?: { config?: Any } }".

File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/dagster/_grpc/impl.py", line 510, in get_external_execution_plan_snapshot
create_execution_plan(
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/dagster/_core/execution/api.py", line 735, in create_execution_plan
resolved_run_config = ResolvedRunConfig.build(job_def, run_config)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/dagster/_core/system_config/objects.py", line 158, in build
raise DagsterInvalidConfigError(

Any advice or examples are super welcome!

[sryza]

It looks like your code example is including config for an op called process_data_for_date, but your job doesn't have an op with that name.

[mathew-mwi]

euggghhhh sorry, yeah, I see how it works now. Sorry, super obvs in hindsight.