Setting the file. One moment.
Subchapter 2.15
references/automation/sensors/basic-sensors.mdMarkdown3 KBView on GitHub
For the basic sensor pattern with cursors, see the main SKILL.md Quick Reference section.
A canonical file sensor that monitors a directory for new files and triggers runs:
import os
import json
import dagster as dg
@dg.sensor(job=my_job, minimum_interval_seconds=30)
def file_sensor(context: dg.SensorEvaluationContext):
# Load cursor (tracks files we've already processed)
processed_files = json.loads(context.cursor) if context.cursor else {}
# Check directory for files
directory = "/data/incoming"
current_files = {}
runs_to_request = []
for filename in os.listdir(directory):
filepath = os.path.join(directory, filename)
mtime = os.path.getmtime(filepath)
current_files[filename] = mtime
# File is new or modified
if filename not in processed_files or processed_files[filename] != mtime:
runs_to_request.append(
dg.RunRequest(
run_key=f"{filename}_{mtime}",
run_config={"ops": {"my_op": {"config": {"filepath": filepath}}}},
)
)
# Update cursor to track current state
return dg.SensorResult(
run_requests=runs_to_request,
cursor=json.dumps(current_files),
)Key pattern: Store file names and modification times in the cursor to track which files have been processed.
Best practices for cursors:
json.dumps() and json.loads() make it easy to store dictionaries or listscontext.cursor is None on first evaluationSensorResult or call context.update_cursor()Two ways to update cursors:
# Option 1: Return SensorResult
return dg.SensorResult(run_requests=[...], cursor=json.dumps(new_state))
# Option 2: Call update_cursor() directly
context.update_cursor(json.dumps(new_state))
yield dg.RunRequest(...)Control evaluation frequency:
@dg.sensor(
job=my_job,
minimum_interval_seconds=60, # Minimum 60 seconds between evaluations
default_status=dg.DefaultSensorStatus.RUNNING, # Auto-enable when deployed
)
def my_sensor(context): ...Important: minimum_interval_seconds is a minimum, not exact. If sensor evaluation takes 10 seconds and the interval is 30 seconds, the next evaluation happens 30 seconds after the previous evaluation started (20 seconds after it completed).
Properties available in sensor context:
cursor: String cursor from the previous evaluation (None if first evaluation)update_cursor(str): Update the cursor for the next evaluationinstance: DagsterInstance for querying the event log or other instance datalog: Logger for recording sensor evaluation detailsrepository_def: Repository containing the sensorresources: Access configured resources (if defined)Example using context.log:
@dg.sensor(job=my_job)
def logging_sensor(context):
context.log.info(f"Evaluating sensor, cursor: {context.cursor}")
# ... sensor logic