编辑“︁Airflow S3KeySensor”︁（章节）

== 代码示例 ==
以下是一个使用 S3KeySensor 的完整示例：

<syntaxhighlight lang="python">
from airflow import DAG
from airflow.providers.amazon.aws.sensors.s3_key import S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    's3_file_sensor_example',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
) as dag:

    wait_for_file = S3KeySensor(
        task_id='wait_for_s3_file',
        bucket_name='my-data-bucket',
        bucket_key='incoming/data_*.csv',
        wildcard_match=True,
        aws_conn_id='aws_default',
        timeout=3600,
        poke_interval=60,
    )

    # 下游任务可以在这里定义
    # process_data = PythonOperator(...)
</syntaxhighlight>

=== 示例说明 ===
1. 传感器会检查 `my-data-bucket` 存储桶中是否存在匹配 `incoming/data_*.csv` 的文件。
2. 使用通配符（`*`）匹配多个文件。
3. 每次检查间隔 60 秒（`poke_interval`）。
4. 如果 3600 秒（1 小时）内未找到文件，任务失败。