Apache Airflow SFTP Provider

Warnings

• breaking Provider version 5.x.x requires Apache Airflow 2.7.0+ (as of March 2026). Older provider versions had lower minimum Airflow requirements (e.g., 2.0.0 requires 2.1.0+, 3.0.0 requires 2.2.0+). Installing a newer provider version on an older Airflow might automatically upgrade Airflow, potentially requiring manual database migration.
  Fix: Ensure your Apache Airflow environment is upgraded to at least version 2.7.0 before installing or upgrading this provider to version 5.x.x. Check the provider's changelog for specific version requirements.
• breaking In `apache-airflow-providers-sftp` version 5.1.0, the `SFTPHook.get_conn()` method no longer directly returns a `paramiko.SFTPClient` instance. Code expecting this specific return type will break.
  Fix: Instead of `sftpHook.get_conn()`, use `sftpHook.get_managed_conn()` when you need a `paramiko.SFTPClient` instance for direct interaction. Alternatively, downgrade the provider to `5.0.0` if direct `get_conn()` usage is critical and refactoring is not immediately feasible.
• breaking For `apache-airflow-providers-sftp` versions 5.0.0 and above, the `apache-airflow-providers-ssh` dependency must be version 4.0.0 or higher. Older versions of the SSH provider will cause issues due to missing keyword arguments.
  Fix: Ensure `pip install apache-airflow-providers-ssh>=4.0.0` is performed alongside or before installing `apache-airflow-providers-sftp` version 5.0.0+.
• deprecated The `timeout` parameter in SFTP connection 'Extra' field is deprecated.
  Fix: Use `conn_timeout` instead of `timeout` for specifying connection timeouts in your SFTP connection 'Extra' parameters.
• deprecated The `private_key_pass` parameter in SFTPHook's connection 'Extra' field was deprecated.
  Fix: Use `private_key_passphrase` instead of `private_key_pass` for consistency with `SSHHook` arguments.
• gotcha The `SFTPOperator` by default does not create intermediate directories on the remote host when transferring files. If the remote path's parent directory does not exist, the operation will fail.
  Fix: Set `create_intermediate_dirs=True` in the `SFTPOperator` to automatically create missing directories in the remote path during file transfer operations.
• gotcha `SFTPOperator` traditionally does not support transferring entire directories, requiring individual file paths. Although an issue was raised for this, it's not a standard feature.
  Fix: For directory transfers, you typically need to list files within the directory (e.g., using `SFTPHook`) and then iterate with `SFTPOperator` for each file, or use an external tool/script.

Install

• pip install apache-airflow-providers-sftp Install core provider

Imports

• SFTPHook

  from airflow.providers.sftp.hooks.sftp import SFTPHook

• SFTPOperator

  from airflow.providers.sftp.operators.sftp import SFTPOperator

• SFTPOperation

  from airflow.providers.sftp.operators.sftp import SFTPOperation

• SFTPSensor

  from airflow.providers.sftp.sensors.sftp import SFTPSensor

Quickstart

This quickstart demonstrates how to use the `SFTPOperator` to upload a file from a local path to an SFTP server and then download it back. It assumes an SFTP connection named `sftp_default` is configured in Airflow. For demonstration purposes, connection details can be provided via environment variables, though typically these are managed within the Airflow UI.

import os
from datetime import datetime

from airflow import DAG
from airflow.providers.sftp.operators.sftp import SFTPOperator, SFTPOperation

# Set default values for connection details via environment variables
SFTP_CONN_ID = os.environ.get('AIRFLOW_CONN_SFTP_DEFAULT', 'sftp_default')
REMOTE_HOST = os.environ.get('SFTP_REMOTE_HOST', 'localhost')
REMOTE_PORT = int(os.environ.get('SFTP_REMOTE_PORT', '22'))
REMOTE_USERNAME = os.environ.get('SFTP_REMOTE_USERNAME', 'sftpuser')
REMOTE_PASSWORD = os.environ.get('SFTP_REMOTE_PASSWORD', 'sftppassword')

# Ensure SFTP_CONN_ID is configured in Airflow UI or via environment variable
# Example for environment variable:
# export AIRFLOW_CONN_SFTP_DEFAULT='sftp://sftpuser:sftppassword@localhost:22/'

with DAG(
    dag_id='sftp_example_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['sftp', 'example', 'file-transfer'],
) as dag:
    upload_file_task = SFTPOperator(
        task_id='upload_local_to_sftp',
        ssh_conn_id=SFTP_CONN_ID,
        local_filepath='/tmp/local_file_to_upload.txt',
        remote_filepath='/tmp/remote_uploaded_file.txt',
        operation=SFTPOperation.PUT,
        create_intermediate_dirs=True,
        # Optional: remote_host=REMOTE_HOST, port=REMOTE_PORT, username=REMOTE_USERNAME, password=REMOTE_PASSWORD
        # These are usually configured in the SFTP_CONN_ID
    )

    download_file_task = SFTPOperator(
        task_id='download_sftp_to_local',
        ssh_conn_id=SFTP_CONN_ID,
        local_filepath='/tmp/local_downloaded_file.txt',
        remote_filepath='/tmp/remote_uploaded_file.txt',
        operation=SFTPOperation.GET,
        # Optional: remote_host=REMOTE_HOST, port=REMOTE_PORT, username=REMOTE_USERNAME, password=REMOTE_PASSWORD
    )

    # For a real scenario, you'd create /tmp/local_file_to_upload.txt before running
    # Example of creating the file:
    # from airflow.operators.python import PythonOperator
    # create_local_file = PythonOperator(
    #    task_id='create_local_file',
    #    python_callable=lambda: open('/tmp/local_file_to_upload.txt', 'w').write('Hello from Airflow!'),
    # )
    # create_local_file >> upload_file_task

    upload_file_task >> download_file_task