来自GoogleCloudStoragePrefixSensor的xcom_pull

from airflow import DAG
from airflow.providers.google.cloud.sensors.gcs import GoogleCloudStoragePrefixSensor
from datetime import datetime, timedelta

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

dag = DAG(
    'gcs_sensor_example',
    default_args=default_args,
    description='A simple GCS sensor example',
    schedule_interval=timedelta(days=1),
)

gcs_sensor_task = GoogleCloudStoragePrefixSensor(
    task_id='gcs_sensor_task',
    bucket='your-gcs-bucket',
    prefix='your-prefix/',
    poke_interval=60,
    timeout=60 * 5,
    dag=dag,
)

from airflow.operators.python_operator import PythonOperator

def process_files(**kwargs):
    # 从XCom或其他方法获取文件列表
    file_list = kwargs['ti'].xcom_pull(task_ids='another_task_that_pushes_file_list')
    
    # 处理文件...
    for file in file_id_list:
        print(f"Processing file: {file}")

process_files_task = PythonOperator(
    task_id='process_files_task',
    python_callable=process_files,
    provide_context=True,
    dag=dag,
)

gcs_sensor_task >> process_files_task