https://dagster.io/ logo
#ask-community
Title
# ask-community
a

Andres Crucetta

04/06/2023, 3:49 PM
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
Copy code
@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.")
👀 1
s

sandy

04/06/2023, 6:31 PM
Hi Andres - sensors don't currently accept config. One way to do this would be to write a function like this:
Copy code
def make_sensor(file_name):
    @sensor(name=f"{file_name}_sensor")
    def _sensor():
        ...
    
    return _sensor
a

Andres Crucetta

04/06/2023, 7:07 PM
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?
s

sandy

04/06/2023, 8:47 PM
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?
👍 1
a

Andres Crucetta

04/07/2023, 4:16 PM
@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
s

sandy

04/07/2023, 4:33 PM
Would I have to build each sensor in this definitions call? With some form of list comprehension
Exactly
10 Views