Hi, is there a good way to organize a sensor that takes a file name and path, and then checks for that file in Azure? I'm trying to just have the file names and paths as configuration for the sensor but it doesn't seem like it works that way. Is there a DRY way to have multiple sensors? I basically have multiple files I want to sense if they exist in Azure, if they do, then I will copy them into Snowflake with an OP. Here's a sample code, I think the context.config_schema doesn't work for sensors which confuses me

@sensor(job=upload_data)
def azure_blob_file_sensor(context):
    """
    Sensor that checks if a file matching a regex pattern exists in a folder in an Azure Blob Storage container.
    """
    regex_pattern: String = context.config_schema["regex_pattern"]
    folder_path: String = context.config_schema["folder_path"]
    time_threshold: int = context.config_schema["time_threshold"]  # Time threshold in seconds

    container_client = ContainerClient.from_connection_string(
        os.environ.get('AZURE_CONNECTION_STRING'), os.environ.get("AZURE_CONTAINER_NAME")
    )

    matched_files = file_exists(container_client, regex_pattern, folder_path)

    if matched_files:
        last_run_time = get_last_run_time(context)
        if last_run_time:
            time_since_last_run = (datetime.utcnow() - last_run_time).total_seconds()
            if time_since_last_run < time_threshold:
                <http://context.log.info|context.log.info>("Job ran recently, skipping execution.")
                return

        for matched_file in matched_files:
            <http://context.log.info|context.log.info>(f"Sensed file: {matched_file.name} at {matched_file.properties.last_modified}")
            yield RunRequest(run_key=matched_file.name)
    else:
        <http://context.log.info|context.log.info>("No matching files found.")

Hi Andres - sensors don't currently accept config. One way to do this would be to write a function like this:

def make_sensor(file_name):
    @sensor(name=f"{file_name}_sensor")
    def _sensor():
        ...
    
    return _sensor

Got it! Is there a DRY way to build multiple sensors for different files? Would you recommend to just split it into multiple functions or does Dagster support that use case?

The code above is fairly DRY - you can invoke

make_sensor

in a for loop on all your files that you want to monitor. Would that work for you?

@sandy, how would you add that code in the definitions? Given that it's not of the expected type

Would I have to build each sensor in this definitions call? With some form of list comprehension

Exactly