Dagster: Example of `@asset_check` with custom `IOManager`

example-asset-check-iomanager.py

USAGE

    $ dagster dev -f example-asset-check-iomanager.py

"""

from dagster import (
    AssetCheckResult,
    Definitions,
    InputContext,
    IOManager,
    OutputContext,
    asset,
    asset_check,
    io_manager
)


class MyIOManager(IOManager):
    def handle_output(self, context: OutputContext, obj):
        pass

    def load_input(self, context: InputContext):
        return "loaded_data"


@io_manager
def my_io_manager():
    return MyIOManager()


@asset(io_manager_key="my_io_manager")
def my_asset():
    return [1, 2, 3]


@asset_check(asset=my_asset)
def my_asset_check(my_asset):
    # `my_asset` is of value `loaded_data` from IOManager
    passed = my_asset == [1, 2, 3]
    return AssetCheckResult(passed=passed)


defs = Definitions(
    assets=[my_asset],
    asset_checks=[my_asset_check],
    resources={"my_io_manager": my_io_manager}
)