Is there a way to trigger an asset materialization from a sensor without defining a job?

[achen187]

I have an asset with dynamic partitions and I want to trigger a partition materialization. To do that, we have to define a job with a partition config as well. Is there a way we can skip defining the job, and just trigger the asset materialization?

I'm imagining something like this, which is an edit of your examples:

images_partitions_def = DynamicPartitionsDefinition(name="images")


@asset(partitions_def=images_partitions_def)
def images(context):
    ...

# asset_key is the asset we want to materialize
@sensor(asset_key="images")
def image_sensor(context):
    new_images = [
        img_filename
        for img_filename in os.listdir(os.getenv("MY_DIRECTORY"))
        # This would instead be referring to whether the asset has the partition, not the job.
        if not context.instance.has_dynamic_partition(
            images_partitions_def.name, img_filename
        )
    ]

    # instead of running a job, I'm request a materialization
    return SensorResult(
        materialization_request=[
            MaterializationRequest(partition_key=img_filename) for img_filename in new_images
        ],
        dynamic_partitions_requests=[
            images_partitions_def.build_add_request(new_images)
        ],
    )

[tacastillo]

Hey Andy!

The sensor decorator takes an asset_selection argument that indicates what assets you want to materialize, similar to the asset_key argument that you're proposing. The RunRequests made will trigger the materializations of the asset passed in.

[achen187]

hm interesting. Can you give me an example of what I'm trying to do above? I'm assuming that syntax I'm trying to pull off doesnt work?

[tacastillo]

Hi!

Yes, the blog post on dynamic partitions has an example, under the How can you use a dynamic partition? heading: https://dagster.io/blog/dynamic-partitioning

@sensor(
   asset_selection=AssetSelection.keys(
       files_metadata.key, processed_files.key, compressed_files.key
   )
)
def file_sensor(context):
    new_files = [
        filename
        for filename in os.listdir(os.getenv("MY_DIRECTORY"))
        if not context.instance.has_dynamic_partition(
            files_partitions_def.name, filename
        )
    ]

    return SensorResult(
        run_requests=[
            RunRequest(partition_key=filename) for filename in new_files
        ],
        dynamic_partitions_requests=[
            files_partitions_def.build_add_request(new_files)
        ],
    )

[jordandakota]

How do you do this where the asset selection is dynamic? Instead of having 10 sensors to monitor 10 tables and materialize them, I can have 1 sensor monitor 10 tables and determine the subset that should be materialized. There's no clear way to get the sensor to trigger just a subset asset_selection.

[garethbrickman]

@jordandakota It's not possible to do that using the asset_selection parameter. In the sensor code itself you would need to build in the logic for deciding the assets and generate RunRequests for them.

[mmutso-boku]

That could be done with a @multi_asset_sensor, I believe?