Airflow Hooks and Plugins

Hooks are Python classes that manage connections to external systems. Plugins let you extend Airflow with custom operators, hooks, sensors, and UI pages. Both tools make Airflow more powerful without modifying its core code.

What Is a Hook?

A Hook wraps the technical details of connecting to an external service. It reads the connection credentials from Airflow's connection store and exposes simple methods you can call from your tasks.

Without a Hook:
──────────────────────────────────────────────────
def upload_to_postgres():
    import psycopg2
    conn = psycopg2.connect(
        host="db.company.com",
        user="admin",
        password="Secret123",
        dbname="sales",
        port=5432
    )
    cursor = conn.cursor()
    cursor.execute("INSERT INTO ...")
    conn.commit()
    conn.close()
── 20+ lines of boilerplate ──

With a Hook:
──────────────────────────────────────────────────
def upload_to_postgres():
    hook = PostgresHook(postgres_conn_id="prod_db")
    hook.run("INSERT INTO ...")
── 2 lines ──

Hooks are reusable. Multiple tasks can use the same hook type — each just passes a different conn_id.

Common Built-In Hooks

  

    

      
Hook Class
      
System It Connects To
      
Provider Package
    
  
  

    

      
PostgresHook
      
PostgreSQL database
      
apache-airflow-providers-postgres
    
    

      
MySqlHook
      
MySQL database
      
apache-airflow-providers-mysql
    
    

      
S3Hook
      
Amazon S3 storage
      
apache-airflow-providers-amazon
    
    

      
GCSHook
      
Google Cloud Storage
      
apache-airflow-providers-google
    
    

      
HttpHook
      
REST APIs over HTTP/HTTPS
      
apache-airflow-providers-http
    
    

      
SFTPHook
      
SFTP file servers
      
apache-airflow-providers-sftp
    
    

      
SlackHook
      
Slack messaging
      
apache-airflow-providers-slack
    
  


Using the PostgresHook
from airflow.providers.postgres.hooks.postgres import PostgresHook

def read_sales_data():
    hook = PostgresHook(postgres_conn_id="prod_sales_db")

    # Run a query and get results as a list of tuples
    records = hook.get_records("SELECT id, amount FROM sales WHERE date = '2024-01-15'")
    for row in records:
        print(f"ID: {row[0]}, Amount: {row[1]}")

def run_insert():
    hook = PostgresHook(postgres_conn_id="prod_sales_db")
    hook.run("INSERT INTO daily_totals VALUES (CURRENT_DATE, 15000)")
Using the S3Hook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

def upload_report():
    hook = S3Hook(aws_conn_id="aws_default")

    hook.load_file(
        filename="/tmp/report.csv",
        key="reports/2024/01/report.csv",
        bucket_name="my-company-reports",
        replace=True,
    )
    print("Report uploaded to S3")

def check_file_exists():
    hook = S3Hook(aws_conn_id="aws_default")
    exists = hook.check_for_key("reports/2024/01/report.csv", "my-company-reports")
    print(f"File exists: {exists}")
Using the HttpHook
from airflow.providers.http.hooks.http import HttpHook

def call_weather_api():
    hook = HttpHook(http_conn_id="weather_api", method="GET")

    # The base URL is stored in the Connection; only the endpoint goes here
    response = hook.run(endpoint="/v1/current?city=London")
    data = response.json()
    print(f"Temperature: {data['temperature']}°C")
Building a Custom Hook
When no built-in hook supports your system, build your own. A custom hook inherits from BaseHook:
from airflow.hooks.base import BaseHook
import requests

class MyAPIHook(BaseHook):
    """Hook for My Custom REST API."""

    conn_name_attr = "my_api_conn_id"
    default_conn_name = "my_api_default"

    def __init__(self, my_api_conn_id="my_api_default"):
        super().__init__()
        self.my_api_conn_id = my_api_conn_id

    def get_conn(self):
        conn = self.get_connection(self.my_api_conn_id)
        base_url = f"https://{conn.host}"
        session = requests.Session()
        session.headers.update({"Authorization": f"Bearer {conn.password}"})
        return session, base_url

    def get_orders(self, date_str):
        session, base_url = self.get_conn()
        response = session.get(f"{base_url}/orders?date={date_str}")
        response.raise_for_status()
        return response.json()
Use this custom hook in a task just like a built-in hook:
def fetch_orders():
    hook = MyAPIHook(my_api_conn_id="my_api_prod")
    orders = hook.get_orders("2024-01-15")
    print(f"Fetched {len(orders)} orders")
What Is an Airflow Plugin?
A Plugin bundles custom code — hooks, operators, sensors, macros, and even custom UI pages — into a single installable package that Airflow discovers automatically.
Plugin vs Direct File Drop:
──────────────────────────────────────────────────────────
Option A: Drop a Python file in ~/airflow/plugins/
  → Airflow auto-discovers it (good for small additions)

Option B: Build a proper pip-installable provider package
  → Better for team sharing and version control
──────────────────────────────────────────────────────────
Creating a Simple Plugin
Create a file at ~/airflow/plugins/my_plugin.py:
from airflow.plugins_manager import AirflowPlugin
from airflow.operators.python import PythonOperator

class GreetOperator(PythonOperator):
    """Custom operator that greets a person."""

    def __init__(self, name, *args, **kwargs):
        self.name = name
        super().__init__(
            python_callable=self._greet,
            *args,
            **kwargs,
        )

    def _greet(self):
        print(f"Hello, {self.name}! Welcome to the pipeline.")

class MyPlugin(AirflowPlugin):
    name = "my_plugin"
    operators = [GreetOperator]
After saving the file, restart the webserver. The GreetOperator is now available across all your DAGs:
from plugins.my_plugin import GreetOperator

greet = GreetOperator(task_id="greet_user", name="Alice")
Hooks vs Operators vs Plugins: Relationship Summary
Plugin (container)
├── Hook    → manages the CONNECTION to an external system
├── Operator → defines the TASK using that hook
└── Sensor  → waits for a CONDITION using that hook

Example:
  PostgresHook     → connects to PostgreSQL
  PostgresOperator → runs SQL using PostgresHook
  SqlSensor        → waits until a SQL query returns a result
Finding Available Providers and Hooks
The Airflow Provider catalog lives at the official Airflow documentation site. Search for your target system (Snowflake, Databricks, dbt, Salesforce, etc.) and follow the installation instructions. Most providers install with a single pip command and immediately unlock their hooks and operators.