Hooks — Low-Level External Connections

Operator → Hook Relationship

graph LR
    OP["PostgresOperator<br/>(High-Level)"] -->|"uses internally"| HOOK["PostgresHook<br/>(Low-Level)"]
    HOOK -->|"manages connection"| DB[("PostgreSQL<br/>Database")]
    OP2["S3CreateObjectOperator<br/>(High-Level)"] -->|"uses internally"| HOOK2["S3Hook<br/>(Low-Level)"]
    HOOK2 -->|"manages connection"| S3["S3<br/>Bucket"]
    style OP fill:#4CAF50,stroke:#388E3C,color:#fff
    style OP2 fill:#4CAF50,stroke:#388E3C,color:#fff
    style HOOK fill:#FF9800,stroke:#F57C00,color:#fff
    style HOOK2 fill:#FF9800,stroke:#F57C00,color:#fff
    style DB fill:#336791,stroke:#264d6e,color:#fff
    style S3 fill:#FF9900,stroke:#cc7a00,color:#fff

When to Use Hooks Directly

You use Hooks directly when:

1. There's no operator for your specific operation
2. You need fine-grained control over the connection
3. You're writing a custom operator

from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.sdk import task

@task()
def custom_data_quality_check():
    """Run a custom quality check using PostgresHook directly."""
    hook = PostgresHook(postgres_conn_id="warehouse")

    # Get a SQLAlchemy connection
    conn = hook.get_conn()
    cursor = conn.cursor()

    cursor.execute("""
        SELECT
            COUNT(*) as total_rows,
            COUNT(DISTINCT customer_id) as unique_customers,
            SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) as null_amounts
        FROM sales
        WHERE sale_date = CURRENT_DATE - 1
    """)

    total, unique, nulls = cursor.fetchone()

    # Data quality assertions
    assert total > 0, "No rows found for yesterday!"
    assert nulls / total < 0.01, f"Too many null amounts: {nulls/total:.2%}"
    assert unique > 100, f"Suspiciously few customers: {unique}"

    return {"total": total, "unique_customers": unique, "null_rate": nulls/total}

Common Hooks

This content is reserved for Premium Members.