The cloud-native orchestrator for the whole development lifecycle, with integrated lineage and observability.

dagster

How can I use: <https://dagster.slack.com/archives/C01U954MEER/p1648739596077989?thread_ts=1648625302.209879&amp;cid=C01U954MEER> the file path as the output of an SFTP sensor as a source asset/input to an asset which eventually retrieves the file and stores it? I get: `DagsterInvalidDefinitionError: Input asset '["file_path"]' for asset 'my_asset' is not produced by any of the provided asset ops and is not one of the provided sources ` when passing the file path. When trying to proceed as in the OP: `context.op_config["filename"]` :  dagster fails with `asset() got an unexpected keyword argument 'config_schema' `. How can I pass the output

one Workaround might be an manual explicit `AssetMaterialization`  from the op (as written above the OP works well).

But I think this is not fully correct - or at least does not feel awesome - as then additional sensors are needed to listen on asset materializations to fire any downstream step - whereas the software-defined-assets should automatically build this lineage graph and mark assets as stale (and if not mistaken also allow to update them).

<https://github.com/geoHeil/dagster-ssh-demo/blob/master/SSH_DEMO/sensors/sftp_sensor_asset_real.py#L97> here you go with an reproducible example