https://dagster.io/ logo
Title
j

Jordan

04/17/2023, 9:01 AM
Hi! I have a job with several assets, when the job runs via the Schedule I don't need any particular config (I use an empty string). Sometimes I need to run the job manually by specifying a config, the config I want to define is always the same for each asset (for example
database_1
). The solution below works fine but I am forced to write the database as many times as there are assets. Is there a more concise method to answer my need that would allow me for example to have a shared config between assets?
:dagster-bot-resolve: 1
@asset(config_schema={"database": str})
def asset_1(context):
    pass

@asset(config_schema={"database": str})
def asset_2(asset_1):
    pass

@asset(config_schema={"database": str})
def asset_3(asset_2):
    pass

job = define_asset_job(
    "my_job",
    selection=[asset_1, asset_2, asset_3],
    config={
        "ops": {
            "asset_1": {"config": {"database": ""}},
            "asset_2": {"config": {"database": ""}},
            "asset_3": {"config": {"database": ""}},
        }
    },
)
v

Vinnie

04/17/2023, 3:09 PM
Maybe use the new
Config
API and set
database_1
as the default?
class MyConfig(dagster.Config):
    database: str = "database_1"

@asset
def asset_1(config: MyConfig):
    pass

@asset
def asset_2(config: MyConfig, asset_1):
    pass

@asset
def asset_3(config: MyConfig, asset_2):
    pass
:rainbow-daggy: 1
j

Jordan

04/17/2023, 4:16 PM
Ah yes indeed, I had not paid attention to the new release. It's a great way to share configurations. However, I don't know if this is a bug or not but it is always the default value that is considered, the resource is not dynamic.
I tested the code below:
class MyConfig(ConfigurableResource):
    database: str

    def get_database(self) -> str:
        return self.database

@asset
def asset_1(context, my_config: MyConfig):
    <http://context.log.info|context.log.info>(my_config.get_database())

defs = Definitions(
    assets=[asset_1],
    resources={
        "my_config": MyConfig(database="database_1"),
    },
)
with the following configuration and I got
database_1
in the log
Update: I used
configure_at_launch()
to solve the above problem