When creating a

RunRequest

from a

@sensor

is there any shortcut assigning a RunConfig to multiple (or all) assets in the job? So far I’m finding that I have to define

config = MyJobConfig(...)
yield RunRequest(run_config=RunConfig({
    'asset1': config,
    'asset2': config,
    'asset3': config,
    etc...
})

If this is an anti-pattern and I shouldn’t be passing config information to each asset like this that would be welcome feedback too. It doesn’t appear that

RunConfig({'all':config'})

works, but is there something similar to that?

The example uses ops, but have you seen our docs on ConfigMapping?

I hadn’t found that specific part of the docs. I’d read the API for ConfigMapping and I couldn’t make heads or tails of it tbh. I can try again and come back with more specific questions

I'm gonna acknowledge that I don't think

ConfigMapping

got an update when we updated our config system. let me see how it parallels.

Okay, if I’m reading that example correctly, IF that example was extended to have multiple ops, then the

RunConfig

returned by

simplified_config

would need to look like:

return RunConfig(
        ops={
            "do_something": DoSomethingConfig(config_param=val.simplified_param),
            "do_something_else": DoSomethingConfig(config_param=val.simplified_param)
        }
    )

Is that right? And then the job config can just be

simplified_config

and it will then actually pass along a larger

RunConfig

that provides a config for each asset.

Ah, you're right. my bad. misread the question. let me see what else we have

And I could leave out the

ops=

part, and it will be inferred to be ops because its the first positional input to

RunConfig

? Thats what it looks like from the RunConfig function signature at least. The examples aren’t totally consistent I think.

return RunConfig(
        {
            "do_something": DoSomethingConfig(config_param=val.simplified_param),
            "do_something_else": DoSomethingConfig(config_param=val.simplified_param)
        }
    )

Yeah, that's what I was looking into...not sure what translates or is relevant.

Its looking like its just easier to write out all the assets at this point. 🤷 . Its adding a fair bit of boilerplate overhead to map it out, and I still have to write them all out anyways. The only benefit was I could define the mapping in my assets.py where the asset names were easily readable. Now i’ll just have to pop back and forth. oh well

You could also just make the

ops

dict argument in a loop by looping over your asset names. Or even in a dict comprehension.

{k:DoSomethingConfig(config_param=val.simplified_param) for k in asset_list}

I'm facing this issue too. I have a sensor that create a run request for either

job_x

or

job_y

. Each of them uses a different (large)

AssetSelection

, and thebut all the assets used in either job take a config object of the same class (with slightly different fields.) Calculation of the

AssetSelection

for each job is easy if I just pick the last one in the chain and call

upstream()

on it, but it's really tedious to spell each set of

AssetKeys

out manually -- and because business requirements around these jobs are changing rapidly, I don't feel that manually spelling the asset names out is a long-term solution. All I want is for every asset in the job to use the exact same config object. I feel like this should be very easy... I find it hard to believe the only solution is to list them literally in the source.

It wasn't quite correct for your scenario

RE values of object: Yes, different. The config object only has one field, which is a

datetime.datetime

. For job X, it gets set to `datetime.now()`For job Y, the calculation is a little complicated.

But all the assets in the job receive the same datetime?

yep

So the difference in config is really just from job to job

That's right

In which case I would use a ConfigurableResource to model this config, that way you only have to set the config in one place on the RunRequest, and not enter it specifically for each asset / op

Okay, thanks very much. I'll go check out the documentation and try that out.

Yeah this is a relatively common pattern for setting "global" / shared configuration across assets / ops in a job.

Looks like this example shows a typical setup: https://docs.dagster.io/concepts/resources#with-ops-and-jobs

Yeah you configure resources at the Definition level, and then Dagster figures out what asset needs what resource based on type hints. I think you'll need to use configure_at_launch to defer actually setting the configuration until your sensor fires

Great, thanks a lot, taht's plenty for me to go on.

No problem, may all your job runs be green and your data error free

It seems as if a datetime is not a valid member for a class that inherits from ConfigurableResource:

Error defining Dagster config class <class 'pipeline.assets.assets.DatetimeResource'> on field 'start_point'.
Unable to resolve config type <class 'datetime.datetime'> to a supported Dagster config type.


This config type can be a:
    - Python primitive type
        - int, float, bool, str, list
    - A Python Dict or List type containing other valid types
    - Custom data classes extending dagster.Config
    - A Pydantic discriminated union type (<https://docs.pydantic.dev/usage/types/#discriminated-unions-aka-tagged-unions>)

Pydantic supports datetimes, but I guess Dagster only allows a subset of what Pydantic supports? That's okay. I'll just make the class member a string instead, and do the conversion from datetime to string at the call site of

RunRequest

Yeah you'll have to use a string. The config types that you put in a Config or ConfigurableResource have to be something that can reasonably be rendered in yaml so that it can be configured in the launchpad. A datetime object has no obvious way to be encoded in a yaml form

ah, gotcha. thanks again

@Maxwell Joslyn In terms of getting a proper datetime-object from your Resource, you can always do this:

class Foo(ConfigurableResource):
  date_str: str
  
  def the_time(self) -> datetime:
    return datetime.strptime(self.date_str, format)
 
@asset
def bar(foo_res: Foo):
  print(foo_res.the_time())

format

could of course be a constant, another field on the Resource, a PrivateAttr, etc., and you could use it to validate date_str before you even start a run.

@DB thats a good point, thanks DB!

I always use resources without Field(), and it works fine (see my post regarding getting a datetime from resource). In fact, I actively avoid setting defaults, because resources with defaults don't make the Launchpad appear when you materialize assets, and I usually do want that to happen. Does your example from above produce an error or not? It looks fine to me (apart from the job name colliding with the asset name, and the missing "Definitions" object, which you need if you have resources).

@DB Thanks for the help, sorry I didn't see it faster. You are right that not setting defaults is useful because of making the launchpad appear! The example above, when furnished with a definitions object, as you say, does indeed produce the error ``dagster._check.ParameterCheckError: Invariant violation for parameter default_value. Description: required arguments should not specify default values` The example above will work if I: 1. use

= Field()

on the

start_point

member var definition. 2. move the config to the job rather than the schedule (with some other shenanigans relating to using a RunConfig object versus using a raw dictionary)