Operators, Sensors, Hooks & Providers

from airflow.providers.standard.operators.python import PythonOperator

extract_task = PythonOperator(
    task_id="extract",           # Unique within DAG
    python_callable=my_function,  # What to execute
    op_kwargs={"param": "value"}, # Arguments to callable
    retries=3,
    retry_delay=timedelta(minutes=2),
    execution_timeout=timedelta(minutes=30),
    pool="database_pool",        # Limit concurrency
)

@task.bash()
def run_dbt() -> str:
    return "dbt run --models staging"

@task.virtualenv(requirements=["pandas==2.0"])
def process_in_venv(data: dict) -> dict:
    import pandas as pd  # Available in virtualenv
    return data

@task.kubernetes(image="python:3.11", namespace="airflow")
def heavy_compute():
    pass

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_file = S3KeySensor(
    task_id="wait_for_file",
    bucket_name="my-bucket",
    bucket_key="incoming/{{ ds }}/data.csv",
    aws_conn_id="aws_default",
    mode="reschedule",           # Release worker slot
    poke_interval=300,           # Check every 5 minutes
    timeout=3600,                # Fail after 1 hour
    exponential_backoff=True,    # Increase interval between checks
    max_wait=timedelta(minutes=30),  # Max interval with backoff
)

from airflow.sensors.base import PokeReturnValue

@task.sensor(mode="reschedule", poke_interval=60, timeout=3600)
def wait_for_api(**context) -> PokeReturnValue:
    from airflow.providers.http.hooks.http import HttpHook
    hook = HttpHook(http_conn_id="api_default")
    response = hook.run(endpoint="/status")
    is_ready = response.json().get("ready", False)
    return PokeReturnValue(is_done=is_ready, xcom_value=response.json())

# Enable deferrable mode globally for all operators that support it
AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True

# Standard (blocks worker)
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

# Deferrable (releases worker) — often same class with deferrable=True
S3KeySensor(
    task_id="wait",
    deferrable=True,  # Uses triggerer instead of worker
    ...
)

@task()
def fetch_from_postgres(**context) -> list[dict]:
    from airflow.providers.postgres.hooks.postgres import PostgresHook
    hook = PostgresHook(postgres_conn_id="my_postgres")
    df = hook.get_pandas_df("SELECT * FROM users WHERE date = %(ds)s", parameters={"ds": context["ds"]})
    return df.to_dict(orient="records")

pip install apache-airflow-providers-amazon
pip install apache-airflow-providers-google
pip install apache-airflow-providers-postgres

from airflow.models import BaseOperator

class MyCustomOperator(BaseOperator):
    template_fields = ("source", "destination")  # Jinja-templatable fields

    def __init__(self, source: str, destination: str, **kwargs):
        super().__init__(**kwargs)
        self.source = source
        self.destination = destination

    def execute(self, context):
        # Task logic here
        self.log.info(f"Processing {self.source} -> {self.destination}")
        return {"status": "success"}

from airflow.hooks.base import BaseHook

class MyApiHook(BaseHook):
    conn_name_attr = "my_api_conn_id"
    default_conn_name = "my_api_default"
    conn_type = "http"

    def __init__(self, my_api_conn_id: str = default_conn_name):
        super().__init__()
        self.my_api_conn_id = my_api_conn_id

    def get_conn(self):
        conn = self.get_connection(self.my_api_conn_id)
        return {"base_url": conn.host, "token": conn.password}

    def fetch_data(self, endpoint: str) -> dict:
        config = self.get_conn()
        # Implementation here
        pass

from airflow.sensors.base import BaseSensorOperator

class MyCustomSensor(BaseSensorOperator):
    def __init__(self, target_value: str, **kwargs):
        super().__init__(**kwargs)
        self.target_value = target_value

    def poke(self, context) -> bool:
        # Return True when condition is met
        current = check_external_system()
        return current == self.target_value

my_airflow_components/
  __init__.py
  operators/
    __init__.py
    my_operator.py
  hooks/
    __init__.py
    my_hook.py
  sensors/
    __init__.py
    my_sensor.py
  setup.py