sensor->graph_backed_asset->asset1->asset2

graph-backed-asset = AssetsDefinition.from_graph(graph)

@asset
def asset1(graph_backed_asset):

@repository
def repo():
    return [
        graph_backed_asset,
        asset1,
        asset2,
        sensor,
        define_asset_job("all_asset_job")
    ]

execute_in_process

dagster._core.errors.DagsterInvalidDefinitionError: Input asset '["graph_backed_asset"]' for asset '["asset1"]' is not produced by any of the provided asset ops and is not one of the provided sources

graph-backed-asset = AssetsDefinition.from_graph(graph)

_

AssetsDefinition.from_graph

@op(
    config_schema={
        "source_bucket_name": str,
        "source_prefix": str,
        "working_folder": str,
    }
)
def download_file(context):
    import os
    from argparse import Namespace

    op_args = Namespace(**context.op_config)

    bucket = storage.Client().bucket(op_args.source_bucket_name)
    blob = bucket.blob(op_args.source_prefix)
    dest_file = os.path.join(
        op_args.working_folder, os.path.basename(op_args.source_prefix)
    )
    blob.download_to_filename(dest_file)
    return dest_file


@graph
def file_download_graph():
    return download_file()


def make_gcp_storage_sensor(
    source_bucket_name="eiq-data-matrix-cil", source_prefix="input/", job=None
):
    import os

    job = job or file_download_graph

    @sensor(
        job=job,
        minimum_interval_seconds=3600,  # every 1 hour
        description=f"Update in {os.path.join(source_bucket_name, source_prefix)} sensed for job {job.name}",
    )
    def gcp_storage_sensor():
        bucket = storage.Client(source_bucket_name).get_bucket(source_bucket_name)

        for blob in bucket.list_blobs(prefix=source_prefix, delimiter="/"):
            if blob.name != source_prefix:
                yield RunRequest(
                    # code omitted
                )

    return gcp_storage_sensor

downloaded_file_asset = AssetsDefinition.from_graph(file_download_graph)

graph_backed_asset