FTP File Availability Monitor

Build a workflow task that monitors an FTP server for file availability and triggers subsequent processing only when files are present.

Requirements

Your task is to implement a monitoring system that waits for specific files to appear on an FTP server before proceeding with downstream workflow tasks.

Core Monitoring

• Monitor a remote FTP path for file or directory presence
• Use polling with configurable intervals to periodically check for existence
• Support timeout configuration to prevent indefinite waiting
• Connect to FTP servers using a connection identifier from the workflow configuration

Protocol Support

• Support both standard FTP and secure FTPS protocols
• Use the appropriate protocol based on the monitoring task type

Dynamic Paths

• Support runtime path resolution using template variables
• Enable paths with workflow parameters (e.g., execution date placeholders)

Test Cases

• Monitoring detects when a file exists at the specified path @test
• Monitoring continues polling when a file does not yet exist @test
• Timeout causes task to complete when file doesn't arrive within time limit @test
• Template variables in paths are resolved correctly at runtime @test

Implementation

@generates

API

"""
FTP File Availability Monitor

Provides monitoring capabilities for FTP file/directory presence.
"""

def create_ftp_file_monitor(
    ftp_conn_id: str,
    path: str,
    timeout: int = 3600,
    poke_interval: int = 60
):
    """
    Create a monitoring task that waits for a file or directory to exist on an FTP server.

    Args:
        ftp_conn_id: Connection ID for the FTP server
        path: Remote path to monitor (supports template variables)
        timeout: Maximum time to wait in seconds (default: 3600)
        poke_interval: Time between checks in seconds (default: 60)

    Returns:
        A monitoring task object that can be used in a workflow
    """
    pass

def create_secure_ftp_file_monitor(
    ftp_conn_id: str,
    path: str,
    timeout: int = 3600,
    poke_interval: int = 60
):
    """
    Create a monitoring task for secure FTPS servers.

    Args:
        ftp_conn_id: Connection ID for the FTPS server
        path: Remote path to monitor (supports template variables)
        timeout: Maximum time to wait in seconds (default: 3600)
        poke_interval: Time between checks in seconds (default: 60)

    Returns:
        A monitoring task object for secure FTP that can be used in a workflow
    """
    pass

Dependencies { .dependencies }

apache-airflow-providers-ftp { .dependency }

Provides FTP operations and monitoring capabilities for Apache Airflow workflows.

@satisfied-by