Depends on D5285.
The main things I'm looking for feedback on are the usage examples in override_asset_store.py and input_asset_store.py. The implementation works, but could use a bunch of cleanup.
- A "manager_key" arg to InputDefinition, which allows users to give a resource that determines how the input is loaded.
- A @input_manager decorator, which defines a resource that loads inputs using the decorated function.
- If an input def has a manager key, it overrides other ways of loading the input. I.e. if input corresponds to an upstream output, the manager will be used instead of the asset store on that upstream output. If the input is unsatisfied, the manager will be used instead of the dagster_type_loader.
- When loading an input that corresponds to an upstream output, the context passed to the loader function includes the step output handle and asset_metadata for the upstream.
- We could consider adding the asset store resource itself to the context, but this might be undesirable: e.g. if the asset store resource constructs a spark session, but the step doing the loading is running in an environment without a spark installation.
- Users can build in flexibility by constructing loaders out of compositions of other loaders.
- It would be easy to provide convenience APIs for building input managers out of dagster_type_loaders or AssetStores.