https://dagster.io/ logo
#ask-community
Title
# ask-community
d

Donny Winston

05/09/2023, 9:58 PM
how do I configure an asset definition for inclusion in a
Definitions
invocation? I remember with graphs, it was
to_job
, and I could pass run config.
🤖 1
s

sean

05/09/2023, 10:01 PM
You can put assets in
Definitions
and access them through an assets job. The default job created by
define_asset_job
will contain all assets in the `Definitions`:
Copy code
defs = Definitions(
    assets=[some_asset],
    jobs=[define_asset_job("all_assets")]
)

defs.get_job_def("all_assets")
d

Donny Winston

05/09/2023, 10:04 PM
sorry, I mean an asset definition that has a
config
param that is a subtype of
dagster.Config
. the example from https://docs.dagster.io/concepts/assets/software-defined-assets#asset-configuration is
Copy code
from dagster import Config, asset


@asset
def my_upstream_asset() -> int:
    return 5


class MyDownstreamAssetConfig(Config):
    api_endpoint: str


@asset
def my_downstream_asset(config: MyDownstreamAssetConfig, my_upstream_asset: int) -> int:
    data = requests.get(f"{config.api_endpoint}/data").json()
    return data["value"] + my_upstream_asset
dagster bot not a thread 1
Copy code
Definitions(
    assets=[my_downstream_asset.configured({"api_endpoint": "foo"})],
?
like I want a set of assets with varying configuration. sort of like it's a partitioned asset I guess, but the various config values wouldn't necessarily result in a set of mutually exclusive and collectively exhaustive assets (i.e. a true "partitioning")
s

sean

05/09/2023, 10:12 PM
I’m not sure I fully understand what you’re looking for, but this works-- does it illustrate what you need?
Copy code
class MyAssetConfig(Config):
    api_endpoint: str


@asset
def my_asset(config: MyAssetConfig) -> str:
    print(config.api_endpoint)
    return config.api_endpoint

defs = Definitions(
    assets=[my_asset],
    jobs=[define_asset_job("all_assets")],
)

defs.get_job_def("all_assets").execute_in_process(
    run_config=RunConfig({"my_asset": MyAssetConfig(api_endpoint="<http://localhost:8080>")})
)
d

Donny Winston

05/09/2023, 10:13 PM
I don't want to
execute_in_process
(or
materialize
) when loading the module. I want to be able to do this e.g. in dagit, but have pre-supplied config.
as it is, I get
Missing required config entry "ops" at the root
in my Dagit launchpad, which makes sense, because required config is missing for an asset materialization.
s

sean

05/09/2023, 10:16 PM
hmmm-- so you want an asset with a particular config schema, but the config params pre-set (but not to the defaults in the schema) when e.g. clicking Materialize in dagit?
you could do this through setting the
config
param on
define_asset_job
, but I’m not sure it’s possible to do this for the hidden base asset job that is used by default when clicking Materialize in dagit
Copy code
defs = Definitions(
    assets=[my_asset],
    jobs=[
        define_asset_job("all_assets", config=RunConfig({"my_asset": MyAssetConfig(api_endpoint="foo")})
    ],
)
d

Donny Winston

05/09/2023, 10:19 PM
hmm, so it sounds like the way to go is to ensure default values for all of an asset's config class?
s

sean

05/09/2023, 10:23 PM
well, let me revise my earlier statement-- if you materialize the assets while viewing the correct job in dagit, then the approach in the above code block should work:
if you click “Materialize” on
my_asset
when looking at the
all_assets
job (but not when looking at
default
asset group), it will use the specified config for
all_assets_job
d

Donny Winston

05/09/2023, 10:25 PM
okay, thanks. and that should still trigger all the asset-y logic hopefully, and
my_asset
materializations will appear in its history regardless of being initiated via a "job" hopefully. thanks.
s

sean

05/09/2023, 10:25 PM
my_asset materializations will appear in its history regardless of being initiated via a “job”
most definitely
👍 1
16 Views