@Scout when I use the input manager how do I specify the input config on an asset so that the input manager can use it?

When using an input manager with assets in Dagster, you can specify the input configuration that the input manager will use by defining an

input_config_schema

on the input manager. This schema defines the expected configuration structure that the input manager will receive. You can then access this configuration within the

load_input

function of the input manager through the

InputContext

object. Here's a step-by-step guide based on the documentation: 1. Define your input manager with an

input_config_schema

. This schema specifies the configuration parameters that your input manager expects. 2. Implement the

load_input

method in your input manager class. Within this method, you can access the configuration via

context.config

. 3. When executing a job that uses this input manager, you can pass the per-input configuration in the

run_config

. Here is an example from the documentation:

python class MyConfigurableInputLoader(InputManager):     def load_input(self, context: InputContext):         return read_dataframe_from_table(name=context.config["table"])  @input_manager(input_config_schema={"table": str}) def my_configurable_input_loader():     return MyConfigurableInputLoader()  # or  @input_manager(input_config_schema={"table": str}) def my_other_configurable_input_loader(context):     return read_dataframe_from_table(name=context.config["table"])

Then, when executing a job, you can pass in the per-input config like this:

python load_table_job.execute_in_process(     run_config={"ops": {"my_op": {"inputs": {"dataframe": {"table": "table_1"}}}}}, )

In this example,

load_table_job

is the job that you're executing,

my_op

is the operation within the job that has an input named

dataframe

, and

table_1

is the value for the

table

parameter that you're passing to the input manager. Remember to replace

my_op

,

dataframe

, and

table_1

with the actual names used in your job definition and the specific configuration you want to pass to your input manager.

@Scout can I also have the table location as part of the source asset metadata or config, so that it's always part of the asset. Then the input manager should pick up the location and pass it into the load function

Yes, you can include metadata as part of the source asset definition, and then access this metadata within your input manager. The metadata for an asset can be defined statically when the asset is declared, and it can be accessed by the input manager when loading the asset as an input to an operation. Here's how you can do it: 1. Define your source asset with metadata that includes the table location:

python from dagster import SourceAsset, MetadataValue  source_asset = SourceAsset(     key=AssetKey("my_table"),     metadata={         "table": MetadataValue.text("my_table_name"),         "schema": MetadataValue.text("my_schema_name"),     } )

2. In your input manager, you can access the metadata of the upstream output that produced the object you're loading. This is done through the

InputContext.upstream_output

property:

python class MyIOManager(IOManager):     def load_input(self, context: InputContext):         if context.upstream_output and context.upstream_output.metadata:             table_name = context.upstream_output.metadata["table"]             schema = context.upstream_output.metadata["schema"]             return read_dataframe_from_table(name=table_name, schema=schema)         else:             raise Exception("Upstream output doesn't have schema and metadata set")

3. When you define your asset that depends on the source asset, you can use the input manager that you've defined:

python from dagster import asset  @asset(io_manager_key="my_io_manager") def my_asset(my_table):     # The my_table parameter is loaded by MyIOManager with the metadata from source_asset     # Perform your operations with my_table here     pass

In this example,

my_io_manager

is the key associated with the

MyIOManager

instance that you've registered in your repository. When

my_asset

is executed, Dagster will use

MyIOManager

to load

my_table

, and the input manager will retrieve the table name and schema from the metadata of the source asset. Make sure to replace

"my_table_name"

,

"my_schema_name"

, and other placeholders with the actual values you intend to use in your asset definitions.

@Scout how do I permanently add the input config on an asset?

In Dagster, you can permanently add configuration to an asset by defining a

config_schema

on the asset itself. This allows you to specify configuration that should be provided when materializing the asset. The configuration can include any parameters that are required for the computation or for loading the asset's inputs. Here's how you can define an asset with a `config_schema`:

python from dagster import asset, Field  @asset(config_schema={"table_name": Field(str, is_required=True)}) def my_asset(context):     table_name = context.op_config["table_name"]     # Use the table_name in your asset computation     ...

In this example,

my_asset

requires a configuration field

table_name

which must be provided when materializing the asset. The

Field

class is used to define the type of the configuration field and whether it is required. When you materialize this asset, you will need to provide the configuration in the `run_config`:

python from dagster import materialize  result = materialize(     [my_asset],     run_config={         "ops": {             "my_asset": {                 "config": {                     "table_name": "my_table"                 }             }         }     } )

The

run_config

specifies the value for

table_name

that the asset will use during execution. If you want to set a default configuration for the asset that can be overridden when needed, you can use the

default_value

parameter of the `Field`:

python @asset(config_schema={"table_name": Field(str, default_value="default_table_name")}) def my_asset_with_default(context):     table_name = context.op_config["table_name"]     # Use the table_name in your asset computation     ...

In this case, if you do not provide a

table_name

in the

run_config

, the asset will use the default value

"default_table_name"

. Remember that the configuration specified in

config_schema

is static and part of the asset's definition. It is not meant to change frequently and is typically used for parameters that are essential to the asset's computation. If you need to change the configuration, you will need to do so in the code where the asset is defined or by providing a different

run_config

when materializing the asset.

@Scout the observable_source_asset doesn't have the option to add a config_schema in the decarator, how do I achieve this now?

As of my knowledge cutoff date in early 2023, the

observable_source_asset

decorator in Dagster does not directly support a

config_schema

parameter. This is because source assets are typically used to represent external data sources and are not computed within Dagster; therefore, they don't have computation logic that would require configuration. However, if you need to associate configuration with a source asset for the purpose of loading it with an input manager or IO manager, you can use the following approaches: 1. Use Metadata: You can attach metadata to the source asset that includes the necessary configuration information. This metadata can then be accessed by the input manager or IO manager when loading the asset.

python
from dagster import SourceAsset, MetadataValue

my_source_asset = SourceAsset(
    key="my_source_asset_key",
    metadata={
        "table_name": MetadataValue.text("my_table_name"),
        # Add other configuration metadata as needed
    }
)

2. Use a Custom Input Manager: You can create a custom input manager that knows how to load the source asset based on predefined logic or external configuration files.

python
from dagster import IOManager, InputContext, io_manager

class MyCustomInputManager(IOManager):
    def load_input(self, context: InputContext):
        # Here you can access the metadata if needed
        table_name = context.upstream_output.metadata.get("table_name")
        # Implement the logic to load the data based on the table_name or other config
        return load_data(table_name)

@io_manager
def my_custom_input_manager():
    return MyCustomInputManager()

3. Use Resources: If the configuration is related to the environment or setup rather than the asset itself, you can define a resource with a

config_schema

and then use this resource within your input manager or IO manager.

python
from dagster import resource, Field

@resource(config_schema={"database_url": Field(str)})
def my_database_resource(init_context):
    return {
        "database_url": init_context.resource_config["database_url"]
    }

# Then, in your IOManager, you can access this resource
class MyIOManager(IOManager):
    def load_input(self, context: InputContext):
        database_url = context.resources.my_database_resource["database_url"]
        # Use the database_url to load the data
        return load_data_from_database(database_url)

When you define your repository, you can then configure the resource with the necessary configuration: ```python from dagster import repository, define_asset_job @repository def my_repository(): return [ my_source_asset, define_asset_job("my_job", selection="*"), my_custom_input_manager, my_database_resource.configured(

@Scout can you show an example on how to access the metadata within the observable function of the observable_source_asset

As of my knowledge cutoff date, the

observable_source_asset

decorator is used to define source assets with observation functions in Dagster. However, the observation function itself does not directly access the metadata of the source asset it is observing. Instead, the observation function is typically used to check for changes in the external data source and return a

DataVersion

object that represents the current version of the data. If you want to access metadata within the observation function, you would typically define that metadata outside of the observation function and access it within the function. Here's an example of how you might do this:

python from dagster import observable_source_asset, DataVersion, MetadataValue  # Define metadata that you want to access in your observation function source_asset_metadata = {     "bucket": MetadataValue.text("my_bucket"),     "key": MetadataValue.text("my_data_key"), }  @observable_source_asset(metadata=source_asset_metadata) def my_source_asset(context):     # Access the metadata from the context     bucket = context.metadata["bucket"].value     key = context.metadata["key"].value      # Use the metadata to perform your observation logic     # For example, check if a file in a cloud storage bucket has changed     data_version = check_for_changes(bucket, key)      # Return a DataVersion object based on the results of your observation     return DataVersion(data_version)

In this example,

check_for_changes

would be a custom function you define that checks the specified bucket and key for changes and returns a string representing the data version. The

DataVersion

object is then created with this string and returned from the observation function. The

context

parameter in the observation function provides access to the metadata defined on the source asset. You can access the metadata values using the

context.metadata

dictionary, where each key corresponds to a metadata entry defined on the source asset. Please note that the specific APIs and capabilities of Dagster may have evolved since my knowledge cutoff date, so I recommend checking the latest Dagster documentation for any updates or changes to the

observable_source_asset

functionality.

@Scout why can't I provide an input manager as IO manager key for an observable source asset: dagster._core.errors.DagsterInvalidDefinitionError: io manager with key 'input_uc_reader' required by SourceAsset with key ["feat_my_feature_update", "source_asset"], but received <class 'dagster._core.storage.input_manager.InputManagerDefinition'>.

The error message you're encountering indicates that you're trying to provide an

InputManagerDefinition

where an

IOManagerDefinition

is expected. In Dagster,

InputManager

and

IOManager

serve different purposes, and they are not interchangeable. - InputManager: An

InputManager

is responsible for loading inputs for ops (operations) or assets. It defines how to retrieve data that an op or asset depends on. - IOManager: An

IOManager

is responsible for handling both the storage of outputs and the retrieval of inputs for ops or assets. It defines how to store outputs and how to retrieve them later as inputs to other ops or assets. When defining a

SourceAsset

, you need to specify an

IOManager

, not an

InputManager

. The

IOManager

is used to load the contents of the asset when it's used as an input to other assets within a job. Here's an example of how you might define a

SourceAsset

with an `IOManager`:

python from dagster import SourceAsset, io_manager, IOManager  class MyIOManager(IOManager):     def handle_output(self, context, obj):         # Logic to store the output         pass      def load_input(self, context):         # Logic to load the input         pass  @io_manager def my_io_manager():     return MyIOManager()  my_source_asset = SourceAsset(     key="my_source_asset_key",     io_manager_key="my_io_manager_key", )

In this example,

MyIOManager

is an

IOManager

that can handle both outputs and inputs. The

my_io_manager

function is a factory that creates instances of

MyIOManager

, and it's decorated with

@io_manager

to indicate that it provides an

IOManager

. When you define your

SourceAsset

, you should use the key associated with the

IOManager

(

"my_io_manager_key"

in this case), not an

InputManager

. If you encounter this error, you should check the following: 1. Ensure that you have defined an

IOManager

and not an

InputManager

for use with your

SourceAsset

. 2. Make sure that the key you provide in the

io_manager_key

argument of the

SourceAsset

corresponds to an

IOManager

registered in your repository. If you have an

InputManager

that you want to use specifically for loading source assets, you will need to convert it to an

IOManager

or wrap it in an

IOManager

that delegates to the

InputManager

for the loading logic. Remember that an

IOManager

must implement both

handle_output

and

load_input

methods, even if the

handle_output

method is a no-op for source assets. Please check the latest Dagster documentation or reach out to the Dagster community for the most up-to-date guidance if there have been changes after my knowledge cutoff date.

@Scout how do I read the metadata on a source_observable_asset?

In Dagster, when you define an

observable_source_asset

, you can attach metadata to it. However, the observation function (the function you decorate with

@observable_source_asset

) does not receive an

AssetExecutionContext

that would allow direct access to the asset's metadata. Instead, it receives an

OpExecutionContext

, which does not have a direct reference to the asset's metadata. If you need to access metadata within the observation function, you would typically define that metadata outside of the observation function and access it within the function. Here's an example of how you might do this:

python from dagster import observable_source_asset, DataVersion, MetadataValue  # Define metadata that you want to access in your observation function source_asset_metadata = {     "bucket": MetadataValue.text("my_bucket"),     "key": MetadataValue.text("my_data_key"), }  @observable_source_asset(metadata=source_asset_metadata) def my_source_asset(context):     # Access the metadata from the context     bucket = context.metadata["bucket"].value     key = context.metadata["key"].value      # Use the metadata to perform your observation logic     # For example, check if a file in a cloud storage bucket has changed     data_version = check_for_changes(bucket, key)      # Return a DataVersion object based on the results of your observation     return DataVersion(data_version)

In this example,

check_for_changes

would be a custom function you define that checks the specified bucket and key for changes and returns a string representing the data version. The

DataVersion

object is then created with this string and returned from the observation function. The

context

parameter in the observation function provides access to the metadata defined on the source asset. You can access the metadata values using the

context.metadata

dictionary, where each key corresponds to a metadata entry defined on the source asset. Please note that the specific APIs and capabilities of Dagster may have evolved since my knowledge cutoff date, so I recommend checking the latest Dagster documentation for any updates or changes to the

observable_source_asset

functionality.

@Scout OpExecutionContext cannot access the asset metadata... AttributeError: 'OpExecutionContext' object has no attribute 'metadata'

I apologize for the confusion. You are correct that

OpExecutionContext

does not have a

metadata

attribute. In Dagster, metadata associated with an asset is not directly accessible within the

OpExecutionContext

of an

observable_source_asset

. If you need to use specific metadata within the observation function of an

observable_source_asset

, you would typically have to manage this metadata outside of the Dagster context and access it within the function using your own mechanisms, such as environment variables, configuration files, or other external sources. Here's a simplified example of how you might manage metadata externally:

python import os from dagster import observable_source_asset, DataVersion  # Assume metadata is set as environment variables bucket = os.getenv("MY_BUCKET") key = os.getenv("MY_DATA_KEY")  @observable_source_asset def my_source_asset(_):     # Use the metadata to perform your observation logic     data_version = check_for_changes(bucket, key)      # Return a DataVersion object based on the results of your observation     return DataVersion(data_version)

In this example,

check_for_changes

is a hypothetical function that checks for changes in the specified bucket and key, and

MY_BUCKET

and

MY_DATA_KEY

are environment variables that you have set with the necessary metadata. Please note that this is a workaround, and the best practice for managing metadata in Dagster may vary depending on your specific use case and the version of Dagster you are using. Always refer to the latest documentation for the most accurate and up-to-date information.