Hey team, I'm looking for clarification on how to ...
# ask-community
Hey team, I'm looking for clarification on how to create a sensor that takes the output of one asset and uses that output as the input to a job. Say I have two assets (A and B) in two different jobs (A and B) and anytime asset A is materialised, I want job B to start in order to materialise asset B (using asset A's output). However I also want to be able to manually trigger job B on occasion. I currently have this, where am I going wrong? The asset sensor documentation is fairly abstracted so doesn't give the specific example
Copy code
def asset_a(config: ObjectClassImUsing):
    <http://logger.info|logger.info>(f'Received input: {config}')
    return updated_object

def asset_b(config: ObjectClassImUsing, asset_a: ObjectClassImUsing):
    <http://logger.info|logger.info>(f'Received input: {config}')
    return updated_more_object

@asset_sensor(asset_key=AssetKey("asset_a"), job=job_b)
def job_b_sensor(context: SensorEvaluationContext, asset_event: EventLogEntry):
    assert asset_event.dagster_event and asset_event.dagster_event.asset_key
    yield RunRequest(
                "asset_b": "What do I put here to kick off a run of job_b with the output of asset_a?"
dagster bot responded by community 1
🤖 1
I suggest you save the output metadata for asset A (path for filesystem iomanagers is being saved already) so you can load it in your sensor code and pass to the job config. This is if you need to do something special with this matadata. As to loading the actual output of A in B - this will already happen, you can use its value with the asset_a parameter you have in your code. You don’t need to set any special config parameters to id it.
Thank you for the response!! What would passing metadata to sensor code look like? As for loading the actual output, are you saying I could leave the RunRequest blank and asset B would obtain the output of asset A automatically?
Hi Ben, Daniel’s suggestions are good, but it sounds like what you are looking for is the Auto-Materialize Daemon. This will let you set a policy on B to materialize whenever A is materialized.
As for loading the actual output, are you saying I could leave the RunRequest blank and asset B would obtain the output of asset A automatically?
If you are using IO managers, yes. In general you do not need to explicitly pass the dependencies of an asset when it is materialized-- you either let the framework do it for you (IO managers), or you load them in the asset itself.
Thanks for the suggestions guys! I've managed to sort passing the output from
to the input of
. But now how do I also manually trigger a run of
by overriding the run config and passing my own values to the input of
This is a bit of a niche API (not common to use run config to stub upstream values) but you can do this for config (that is launchpad YAML, but can be straightforwardly translated to a
Copy code
        value: foobar
That doesn't seem to work, do you have the link to the documentation for that API?
hmm possible I screwed something up, let me test locally. Here is docs: https://docs.dagster.io/concepts/io-management/unconnected-inputs#unconnected-inputs
Ah you know-- it looks like this doesn’t work for assets. Apologies for the wrong advice. I don’t think there is a first-class way to override asset inputs with config-- like I said, it’s an uncommon thing to want to do. However, you could manage it with run config:
Copy code
from typing import Optional

from dagster import Config, asset

def asset_a() -> str:
    return "foo"

class AssetBConfig(Config):
    asset_a: Optional[str] = None

def asset_b(asset_a: str, config: AssetBConfig) -> str:
    asset_a_value = config.asset_a or asset_a
    return asset_a_value + "bar"
🌈 1
Thank you for the detailed response, it's really handy 😁