Operators, Sensors & Hooks In Depth

Chapter 3 named three ways to write a task. This chapter looks at each one in detail, and at the hook, which sits under them:

You will also build your own operator, from the file to the import line. Every output here comes from a real run with Airflow 3.1.

Five operators cover most of the code you will read. In Airflow 3, they come from provider packages:

A provider package holds operators and hooks for one system, such as Postgres or AWS. Look for one before you write your own code for a system.

Some arguments accept Jinja, such as {{ ds }}, the day of the run. These arguments are the template fields of the operator. In our install they were:

Jinja in any other argument stays as plain text, braces and all. Chapter 8 explains templates in full.

Build an operator when the same block of code is in three or more DAGs, and only some values differ. For one use, a @task function is shorter.

Suppose many DAGs count the rows of a table for one day. Here is the operator. It lives in plugins/shop/operators.py:

from airflow.sdk import BaseOperator


class CountRowsOperator(BaseOperator):
    template_fields = ("day",)

    def __init__(self, table: str, day: str, **kwargs):
        super().__init__(**kwargs)
        self.table = table
        self.day = day

    def execute(self, context):
        print(f"counting {self.table} for {self.day}")
        return 6

The class has three parts, and each has one job:

Airflow puts the dags/, plugins/ and config/ folders on the Python path. So a DAG imports the operator like any Python module, with no extra setup:

from shop.operators import CountRowsOperator

count_orders = CountRowsOperator(
    task_id="count_orders",
    table="orders",
    day="{{ ds }}",
)

To share an operator across many Airflow deployments, make it a real Python package, and install it in each image.

Airflow builds every operator each time it parses the DAG file, by default every 30 seconds (chapter 3). So __init__ runs all day, even when no task runs.

That gives two rules. First, no database calls or API calls in __init__, because they would run on every parse. Second, do not change a template field in __init__:

class BadDayOperator(BaseOperator):
    template_fields = ("day",)

    def __init__(self, day: str, **kwargs):
        super().__init__(**kwargs)
        self.day = day.replace("-", "")   # wrong place

    def execute(self, context):
        print(f"day is {self.day}")

In __init__, day is still the text {{ ds }}, which has no dashes to remove. Later, Airflow renders the template and writes the result onto self.day. Your change is lost.

So store template fields exactly as they arrive. Do the change at the start of execute, where the value is already rendered.

A sensor checks something again and again, such as whether a file exists. Between checks, it waits. How it waits is set by its mode:

Here is one wait for the orders file, written in each mode:

# poke: holds a worker slot until the file comes
S3KeySensor(task_id="wait", bucket_key="s3://raw/{{ ds }}/orders.csv")

# reschedule: checks every 5 minutes, frees the slot in between
S3KeySensor(task_id="wait", bucket_key="s3://raw/{{ ds }}/orders.csv",
            mode="reschedule", poke_interval=300, timeout=6 * 60 * 60)

# deferrable: the triggerer waits, no worker slot at all
S3KeySensor(task_id="wait", bucket_key="s3://raw/{{ ds }}/orders.csv",
            deferrable=True, timeout=6 * 60 * 60)

Suppose 40 poke sensors wait for files, and you have 32 worker slots. Every slot holds a sleeping sensor, and no real work can start.

The workers look busy, so people add workers. But the new slots fill with sensors too. Changing the mode fixes it, with no new machines.

By default, a sensor gives up after 604800 seconds, which is 7 days. For a daily file, that is far too long. The runs behind it pile up, and nobody is told.

So set timeout on every sensor. Base it on when the business needs the data, not on the technical wait.

When the timeout ends, soft_fail decides what happens. We ran a sensor that never succeeds, with timeout=3, in both ways:

A failed sensor is easy to see, and failure alerts fire. A skipped sensor skips everything after it, with no failure anywhere. The run does nothing, and nobody is told.

Use soft_fail=True only when a missing file is normal. For example, a partner sends a file only on some days. Write that reason in a comment.

A hook is a client for one outside system. It finds the connection by its conn_id, logs in, and gives you methods such as get_records. Chapter 9 explains connections.

Most operators are a thin layer around a hook. So inside a @task, you can use the hook yourself, and add your own checks:

# Operator: short, when one query is all you need
SQLExecuteQueryOperator(
    task_id="count_orders",
    conn_id="shop_db",
    sql="SELECT count(*) FROM orders WHERE order_date = '{{ ds }}'",
)

# Hook in a @task: when you need logic around the query
@task
def count_orders(ds=None):
    hook = PostgresHook(postgres_conn_id="shop_db")
    count = hook.get_first(
        "SELECT count(*) FROM orders WHERE order_date = %s", parameters=(ds,)
    )[0]
    if count == 0:
        raise ValueError(f"no orders for {ds}")
    return count

The operator cannot stop the run on zero orders. The hook version can, because it is ordinary Python.

SQL hooks share the same methods:

Choose like this:

• One standard step: an operator.
• Logic around a call: a hook in a @task.
• The same wrapper in three or more DAGs: your own operator.

Operators do, sensors wait, and hooks connect:

1. Look for a provider package before you write your own integration.
2. Your own operator: subclass BaseOperator, then super().__init__ first, store arguments, and work in execute.
3. Put it in plugins/, and import it like any module.
4. Never change a template field in __init__. Change it in execute.
5. For long waits, use reschedule or deferrable, and always set timeout.

Check yourself: you set self.table = table.upper() in __init__, and table is a template field. What does execute see? (Answer: the rendered value, not upper case.)