Sensor to trigger materialization of partitioned dbt assets

[fredonia88]

I have a sensor that queries a redshift table for a new date based on a daily partition definition. If the date is found, it triggers a run request for a dbt job that materializes two dbt partitioned assets.

@sensor(
        name='my_sensor',
        minimum_interval_seconds=3600,
        job=my_dbt_job
    )
    def my_partitioned_job_sensor(context: SensorEvaluationContext, redshift: RedshiftClientResource):
        partition_date = str(my_job.partitions_def.get_last_partition_key())
        format_args = {'partition_date': partition_date}
        query = f"select 1 from my_table where my_date = '{partition_date}';"
        table_is_ready = redshift.get_client().execute_queries(query, fetch_results=True)[0]

        job_name = job.name
        if table_is_ready:
            yield RunRequest(
                run_key=f'run_for_{job_name}_{partition_date}',
                partition_key=f'{partition_date}'
                )

However, a partition in my redshift table may not be available until days later, after other partitions have become available. How can I have my sensor continue to look for an older partitions after new ones become available? I could select all partitions, but how can I determine which of those have been already materialized?

For example, the sensor will check for the 6/8 partition until get_last_partition_key becomes 6/9. If both 6/8 and 6/9 are still unavailable, how can I have this sensor continue to monitor for both 6/8 and 6/9? And furthermore, how can I have the sensor continue to look for 6/8 after 6/9 has materialized?