Airflow hooks

A hook is an abstraction of a specific API that allows Airflow to interact with an external system. Hooks are built into many operators, but they can also be used directly in DAG code.

In this guide, you’ll learn about using hooks in Airflow and when you should use them directly in DAG code. You’ll also implement two different hooks in a DAG.

Over 200 hooks are available in the Airflow Registry. If a hook isn’t available for your use case, you can write your own and share it with the community.

Assumed knowledge

To get the most out of this guide, you should have an understanding of:

• Basic Airflow concepts. See Introduction to Apache Airflow.
• Basic Python. See the Python Documentation.

Hook basics

Hooks wrap around APIs and provide methods to interact with different external systems. Hooks standardize how Astronomer interacts with external systems and using them makes your DAG code cleaner, easier to read, and less prone to errors.

To use a hook, you typically only need a connection ID to connect with an external system. For more information about setting up connections, see Manage your connections in Apache Airflow.

All hooks inherit from the BaseHook class, which contains the logic to set up an external connection with a connection ID. On top of making the connection to an external system, individual hooks can contain additional methods to perform various actions within the external system. These methods might rely on different Python libraries for these interactions. For example, the S3Hook relies on the boto3 library to manage its Amazon S3 connection.

The S3Hook contains over 20 methods to interact with Amazon S3 buckets. The following are some of the methods that are included with S3Hook:

• check_for_bucket: Checks if a bucket with a specific name exists.
• list_prefixes: Lists prefixes in a bucket according to specified parameters.
• list_keys: Lists keys in a bucket according to specified parameters.
• load_file: Loads a local file to Amazon S3.
• download_file: Downloads a file from the Amazon S3 location to the local file system.

When to use hooks

Since hooks are the building blocks of operators, their use in Airflow is often abstracted away from the DAG author. However, there are some cases when you should use hooks directly in a Python function in your DAG. The following are some general guidelines for using hooks in Airflow:

• Hooks should always be used over manual API interaction to connect to external systems.
• If you write a custom operator to interact with an external system, it should use a hook.
• When an operator with built-in hooks exists for your specific use case, you should use the operator instead of manually setting up a hook.
• If you regularly need to connect to an API and a hook is not available, write your own hook and share it with the community.

Example implementation

The following example shows how you can use the hooks (S3Hook and SlackHook) to retrieve values from files in an Amazon S3 bucket, run a check on them, post the result of the check on Slack, and then log the response of the Slack API.

For this use case, you’ll use hooks directly in your Python functions because none of the existing Amazon S3 operators can read data from multiple files within an Amazon S3 bucket. Also, none of the existing Slack operators can return the response of a Slack API call, which you might want to log for monitoring purposes.

The source code for the hooks used in this example can be found in the following locations:

• S3Hook source code
• SlackHook source code

Prerequisites

Before running the example DAG, make sure you have the necessary Airflow providers installed. If you are using the Astro CLI, add the following packages to your requirements.txt file:

apache-airflow-providers-amazon
apache-airflow-providers-slack

Create the connections

1. In the Airflow UI, go to Admin > Connections and click the plus (+) icon to add a new connection.
2. In the Connection Id field, enter a unique name for the connection.
3. In the Connection Type list, select Amazon S3 as the connection type for the Amazon S3 bucket. If the Amazon S3 connection type isn’t available, make sure you installed the provider correctly.
4. Enter your AWS access key ID in the Login field.
5. Enter your AWS secret access key in the Password field. To retrieve your AWS access key ID and AWS secret access key, see AWS Account and Access Keys.
6. Click Save.
7. Repeat steps 1 to 6 to create a new connection for Slack. Select Slack Webhook as the connection type and enter your Bot User OAuth Token in the Password field. To obtain the token, go to Features > OAuth & Permissions on api.slack.com/apps.

Run the example DAG

The following example DAG uses Airflow Decorators to define tasks and XCom to pass information between Amazon S3 and Slack. The name of the Amazon S3 bucket and the names of the files that the first task reads are stored as environment variables for security purposes.

The following example DAG completes the following steps:

• A Python task with a manually implemented S3Hook reads three specific keys from Amazon S3 with the read_key method and then returns a dictionary with the file contents converted to integers.
• A second Python task completes a simple sum check using the results from the first task.
• The SlackHook call method posts the sum check results to a Slack channel and returns the response from the Slack API.

1
# importing necessary packages
2
from datetime import datetime
3
from airflow.decorators import dag, task
4
from airflow.providers.slack.hooks.slack import SlackHook
5
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
6
7
# set bucket name and file names
8
S3BUCKET_NAME = "myhooktutorial"
9
S3_EXAMPLE_FILE_NAME_1 = "file1.txt"
10
S3_EXAMPLE_FILE_NAME_2 = "file2.txt"
11
S3_EXAMPLE_FILE_NAME_3 = "file3.txt"
12
13
14
# task to read 3 keys from your S3 bucket
15
@task
16
def read_keys_from_s3():
17
    s3_hook = S3Hook(aws_conn_id="aws_conn")
18
    response_file_1 = s3_hook.read_key(
19
        key=S3_EXAMPLE_FILE_NAME_1, bucket_name=S3BUCKET_NAME
20
    )
21
    response_file_2 = s3_hook.read_key(
22
        key=S3_EXAMPLE_FILE_NAME_2, bucket_name=S3BUCKET_NAME
23
    )
24
    response_file_3 = s3_hook.read_key(
25
        key=S3_EXAMPLE_FILE_NAME_3, bucket_name=S3BUCKET_NAME
26
    )
27
28
    response = {
29
        "num1": int(response_file_1),
30
        "num2": int(response_file_2),
31
        "num3": int(response_file_3),
32
    }
33
34
    return response
35
36
37
# task running a check on the data retrieved from your S3 bucket
38
@task
39
def run_sum_check(response):
40
    if response["num1"] + response["num2"] == response["num3"]:
41
        return (True, response["num3"])
42
    return (False, response["num3"])
43
44
45
# task posting to slack depending on the outcome of the above check
46
# and returning the server response
47
@task
48
def post_to_slack(sum_check_result):
49
    slack_hook = SlackHook(slack_conn_id="hook_tutorial_slack_conn")
50
51
    if sum_check_result[0] is True:
52
        server_response = slack_hook.call(
53
            api_method="chat.postMessage",
54
            json={
55
                "channel": "#test-airflow",
56
                "text": f"""All is well in your bucket!
57
                        Correct sum: {sum_check_result[1]}!""",
58
            },
59
        )
60
    else:
61
        server_response = slack_hook.call(
62
            api_method="chat.postMessage",
63
            json={
64
                "channel": "#test-airflow",
65
                "text": f"""A test on your bucket contents failed!
66
                        Target sum not reached: {sum_check_result[1]}""",
67
            },
68
        )
69
70
    # return the response of the API call (for logging or use downstream)
71
    return server_response
72
73
74
# implementing the DAG
75
@dag(
76
    dag_id="hook_tutorial",
77
    start_date=datetime(2022, 5, 20),
78
    schedule="@daily",
79
    catchup=False,
80
)
81
def hook_tutorial():
82
    # the dependencies are automatically set by XCom
83
    response = read_keys_from_s3()
84
    sum_check_result = run_sum_check(response)
85
    post_to_slack(sum_check_result)
86
87
88
hook_tutorial()

1
# importing necessary packages
2
from datetime import datetime
3
from airflow import DAG
4
from airflow.operators.python import PythonOperator
5
from airflow.providers.slack.hooks.slack import SlackHook
6
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
7
8
# set bucket name and file names
9
S3BUCKET_NAME = "myhooktutorial"
10
S3_EXAMPLE_FILE_NAME_1 = "file1.txt"
11
S3_EXAMPLE_FILE_NAME_2 = "file2.txt"
12
S3_EXAMPLE_FILE_NAME_3 = "file3.txt"
13
14
15
# function to read 3 keys from your S3 bucket
16
def read_keys_from_s3_function():
17
    s3_hook = S3Hook(aws_conn_id="aws_conn")
18
    response_file_1 = s3_hook.read_key(
19
        key=S3_EXAMPLE_FILE_NAME_1, bucket_name=S3BUCKET_NAME
20
    )
21
    response_file_2 = s3_hook.read_key(
22
        key=S3_EXAMPLE_FILE_NAME_2, bucket_name=S3BUCKET_NAME
23
    )
24
    response_file_3 = s3_hook.read_key(
25
        key=S3_EXAMPLE_FILE_NAME_3, bucket_name=S3BUCKET_NAME
26
    )
27
28
    response = {
29
        "num1": int(response_file_1),
30
        "num2": int(response_file_2),
31
        "num3": int(response_file_3),
32
    }
33
34
    return response
35
36
37
# function running a check on the data retrieved from your S3 bucket
38
def run_sum_check_function(response):
39
    if response["num1"] + response["num2"] == response["num3"]:
40
        return (True, response["num3"])
41
    return (False, response["num3"])
42
43
44
# function posting to slack depending on the outcome of the above check
45
# and returning the server response
46
def post_to_slack_function(sum_check_result):
47
    slack_hook = SlackHook(slack_conn_id="hook_tutorial_slack_conn")
48
49
    if sum_check_result[0] is True:
50
        server_response = slack_hook.call(
51
            api_method="chat.postMessage",
52
            json={
53
                "channel": "#test-airflow",
54
                "text": f"""All is well in your bucket!
55
                        Correct sum: {sum_check_result[1]}!""",
56
            },
57
        )
58
    else:
59
        server_response = slack_hook.call(
60
            api_method="chat.postMessage",
61
            json={
62
                "channel": "#test-airflow",
63
                "text": f"""A test on your bucket contents failed!
64
                        Target sum not reached: {sum_check_result[1]}""",
65
            },
66
        )
67
68
    # return the response of the API call (for logging or use downstream)
69
    return server_response
70
71
72
# implementing the DAG
73
with DAG(
74
    dag_id="hook_tutorial",
75
    start_date=datetime(2022, 5, 20),
76
    schedule="@daily",
77
    catchup=False,
78
    # Render templates using Jinja NativeEnvironment
79
    render_template_as_native_obj=True,
80
):
81
    read_keys_form_s3 = PythonOperator(
82
        task_id="read_keys_form_s3", python_callable=read_keys_from_s3_function
83
    )
84
85
    run_sum_check = PythonOperator(
86
        task_id="run_sum_check",
87
        python_callable=run_sum_check_function,
88
        op_kwargs={
89
            "response": "{{ ti.xcom_pull(task_ids='read_keys_form_s3', \
90
                key='return_value') }}"
91
        },
92
    )
93
94
    post_to_slack = PythonOperator(
95
        task_id="post_to_slack",
96
        python_callable=post_to_slack_function,
97
        op_kwargs={
98
            "sum_check_result": "{{ ti.xcom_pull(task_ids='run_sum_check', \
99
                key='return_value') }}"
100
        },
101
    )
102
103
    # the dependencies are automatically set by XCom
104
    read_keys_form_s3 >> run_sum_check >> post_to_slack