@Scout

from dagster import asset, Config

class MyAssetConfig(Config):
    person_name: str


class MyAssetConfigDefault(Config):
    person_name: str = "default name"

@asset(group_name="playground")
def greeting(context, config: MyAssetConfig) -> str:
    <http://context.log.info|context.log.info>(f"Hello {config.person_name}")
    return f"hello {config.person_name}"


def myfactory(name, cfg=None):
    if cfg is None:
        @asset(group_name="playground", name=name)
        def inner_asset(context, config: MyAssetConfig) -> str:
            <http://context.log.info|context.log.info>(f"Hello from factory {config.person_name}")
            return f"hello from factory{config.person_name}"
    else:
        @asset(group_name="playground", name=name)
        def inner_asset(context, config: cfg) -> str:
            <http://context.log.info|context.log.info>(f"Hello from factory {config.person_name}")
            return f"hello from factory{config.person_name}"
    return inner_asset

greeting_factory = myfactory("greeting_factory")

greeting_factory_config = myfactory("greeting_factory_config", cfg=MyAssetConfigDefault)

is already solving my needs almost 95% I would want to specify job job1 for asset a where the default value of the config is chosen but job_2 where for asset a some other value is chosen from the config. 0 replies

Based on the documentation, you can define assets with configuration schemas using the

Config

class from Dagster. You can also specify default values for config fields within these classes. In your code, you have defined two configuration classes,

MyAssetConfig

and

MyAssetConfigDefault

, the latter having a default value for

person_name

. To specify different configurations for different jobs, you can define jobs that materialize your assets with specific run configurations. Here's an example of how you might define two jobs that use the

greeting

asset with different configurations:

python from dagster import job, RunConfig  # Define a job that uses the default config for the greeting asset @job def job1():     greeting()  # Define a job that specifies a different config for the greeting asset @job def job2():     greeting.with_config({"person_name": "some other name"})  # Materialize job1 with the default config result_job1 = job1.execute_in_process()  # Materialize job2 with a specific config result_job2 = job2.execute_in_process()

In this example,

job1

uses the default configuration for the

greeting

asset, while

job2

specifies a different value for

person_name

. When you materialize these jobs, they will use the respective configurations you've defined. Please note that the code provided above is a conceptual example based on the documentation's guidance on how to use configuration schemas with assets and jobs. You may need to adjust the code to fit the exact structure and requirements of your Dagster project.

@Scout from dagster import asset, Config class MyAssetConfig(Config): person_name: str class MyAssetConfigDefault(Config): person_name: str = "default name" @asset(group_name="playground") def greeting(context, config: MyAssetConfig) -> str: context.log.info(f"Hello {config.person_name}") return f"hello {config.person_name}" def myfactory(name, cfg=None): if cfg is None: @asset(group_name="playground", name=name) def inner_asset(context, config: MyAssetConfig) -> str: context.log.info(f"Hello from factory {config.person_name}") return f"hello from factory{config.person_name}" else: @asset(group_name="playground", name=name) def inner_asset(context, config: cfg) -> str: context.log.info(f"Hello from factory {config.person_name}") return f"hello from factory{config.person_name}" return inner_asset greeting_factory = myfactory("greeting_factory") greeting_factory_config = myfactory("greeting_factory_config", cfg=MyAssetConfigDefault) is already solving my needs almost 95% I would want to specify job job1 for asset a where the default value of the config is chosen but job_2 where for asset a some other value is chosen from the config. However, this fails with: dagster._check.ParameterCheckError: Param "config" is not one of ['ConfigMapping', 'Mapping', 'PartitionedConfig', 'RunConfig']. Got MyAssetConfig(person_name='job config 1') which is type <class 'ascii.ascii.assets.playground.config_mini.MyAssetConfig'>. for from dagster import define_asset_job greeting_job1 = define_asset_job( "greeting_job1", selection=[greeting_factory_config], config=MyAssetConfig(person_name="job config 1"), ) greeting_job2 = define_asset_job( "greeting_job2", selection=[greeting_factory_config], config=MyAssetConfigDefault(person_name="222 default"), ) from dagster import Definitions defs = Definitions( assets=[greeting, greeting_factory, greeting_factory_config], jobs=[greeting_job2, greeting_job1], ) how can this be fixed

The error you're encountering indicates that the

config

parameter you're passing to

define_asset_job

is not of an expected type. In Dagster, when you define a job and want to provide configuration for the assets within that job, you should use a dictionary to specify the configuration for each asset, rather than passing an instance of a

Config

subclass directly. Here's how you can fix the issue by providing the configuration as a dictionary:

python from dagster import define_asset_job  # Define job1 with a specific config for the greeting_factory_config asset greeting_job1 = define_asset_job(     "greeting_job1",     selection=[greeting_factory_config],     config={         "ops": {             "greeting_factory_config": {                 "config": {"person_name": "job config 1"}             }         }     }, )  # Define job2 with a different config for the greeting_factory_config asset greeting_job2 = define_asset_job(     "greeting_job2",     selection=[greeting_factory_config],     config={         "ops": {             "greeting_factory_config": {                 "config": {"person_name": "222 default"}             }         }     }, )  from dagster import Definitions defs = Definitions(     assets=[greeting, greeting_factory, greeting_factory_config],     jobs=[greeting_job1, greeting_job2], )

In this corrected code, the

config

parameter for each

define_asset_job

call is a dictionary that specifies the configuration for the

greeting_factory_config

asset. The key

"ops"

is used to specify configurations for each asset within the job, and the asset name is used as a key within the

"ops"

dictionary to provide the specific configuration for that asset. Please note that the exact structure of the configuration dictionary may vary depending on the version of Dagster you are using and the specific requirements of your assets and jobs. Always refer to the latest Dagster documentation for the most accurate and up-to-date information on defining jobs and asset configurations.