Process a file from S3 only if it changed since the last execution

Reprocess an Amazon S3 object only when it has actually changed, instead of blindly running on every schedule tick. This flow checks the S3 object's LastModified timestamp against the previous execution time and skips work when the file is unchanged, saving compute and avoiding duplicate downstream processing for incremental S3 data pipelines.

How it works

• The schedule trigger (io.kestra.plugin.core.trigger.Schedule) fires every five minutes via the cron */5 * * * *.
• The process_file_if_changed task (io.kestra.plugin.scripts.python.Commands) runs a Python script in a Docker taskRunner (io.kestra.plugin.scripts.runner.docker.Docker), installing awswrangler as a dependency.
• The task loads s3_modified.py from namespaceFiles and calls it with the bucket and object variables plus {{ trigger.date ?? execution.startDate }} as the comparison date.
• The script uses boto3 head_object to read the S3 object's LastModified value and compares it to the passed-in date, printing whether the file needs reprocessing or is unchanged.

Add the following Python script named s3_modified.py in the Editor:

import boto3
from datetime import datetime
import argparse

def parse_date(date_str):
    if date_str.endswith('Z'):
        return datetime.fromisoformat(date_str.replace('Z', '+00:00'))
    return datetime.fromisoformat(date_str)

def check_s3_object_modification(bucket, object, trigger_date):
    s3 = boto3.client("s3")
    response = s3.head_object(Bucket=bucket, Key=object)
    last_modified = response["LastModified"]
    comparison_datetime = parse_date(trigger_date)

    if last_modified > comparison_datetime:
        print(f"The file '{object}' was modified at {last_modified} and needs to be reprocessed.")
    else:
        print(f"The file '{object}' is unchanged.")

def main():
    parser = argparse.ArgumentParser(description='Check if an S3 object was modified after a given date.')
    parser.add_argument('bucket_name', help='Name of the S3 bucket')
    parser.add_argument('object_key', help='Key of the S3 object')
    parser.add_argument('comparison_date', help='Date to compare against in ISO format')

    args = parser.parse_args()

    check_s3_object_modification(args.bucket_name, args.object_key, args.comparison_date)

if __name__ == "__main__":
    main()

What you get

• Conditional reprocessing that only acts when the S3 object actually changed.
• A reusable comparison date derived from the trigger time, with a fallback to the execution start date.
• Isolated, reproducible runs in a Docker container with awswrangler and boto3 available.
• A clear log line for both the changed and unchanged cases.

Who it's for

• Data engineers building incremental ingestion from S3.
• Platform teams that want to avoid redundant downstream jobs.
• Anyone polling object stores for change detection without a dedicated event pipeline.

Why orchestrate this with Kestra

A bare cron job would reprocess the file on every tick regardless of whether it changed. Kestra adds the event-aware comparison ({{ trigger.date ?? execution.startDate }}), declarative YAML, retries, execution history, and lineage across runs. The Schedule trigger and namespace files let you keep the logic versioned and observable, filling the gap that a plain scheduler and ad hoc scripts cannot: knowing what changed, when, and what ran as a result.

Prerequisites

• A Kestra instance with Docker available for the task runner.
• An S3 bucket and object you can read.

Secrets

• AWS_ACCESS_KEY_ID
• AWS_SECRET_ACCESS_KEY
• AWS_DEFAULT_REGION

Quick start

1. Add the s3_modified.py script as a namespace file in the Editor.
2. Add the AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_DEFAULT_REGION Secrets.
3. Set the bucket and object variables to point at your S3 object.
4. Save the flow and let the schedule run, or trigger it manually.

How to extend

• Replace the print statement with real processing: download the object and chain Kestra tasks to transform or load it.
• Swap the Schedule trigger for an S3 event trigger if you prefer push over polling.
• Loop over multiple objects or a prefix to monitor a whole folder.
• Add notifications on change detection via a Slack or email task.

Links

• Python script plugin docs
• AWS plugin docs
• AWS orchestration with Kestra