https://dagster.io/ logo
#ask-ai
Title
# ask-ai
g

geoHeil

02/28/2024, 10:51 AM
@Scout
Copy code
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
s

Scout

02/28/2024, 10:51 AM
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:
Copy code
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.
g

geoHeil

02/28/2024, 11:02 AM
@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
s

Scout

02/28/2024, 11:02 AM
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:
Copy code
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.
2 Views