Suppose a task downloads data from an HTTP API and uploads it to MinIO. Both systems need connection details, but putting credentials directly in the DAG makes them harder to manage securely. An Airflow connection gives each system an ID that a hook, operator, or custom Python function can use to retrieve its configuration at run time. section 13.3.1 section 12.1.2
For a custom HTTP hook, the setup is:
- Create a connection under Admin > Connections. Give it an ID, choose HTTP, and enter the host, login, password, port, and optional schema. section 9.2.1
- Pass that ID to the hook. In get_conn(), call self.get_connection(self._conn_id); use the returned configuration’s schema, host, and port to form the base URL, and its login and password to authenticate the requests session. section 9.2.1
- Keep defaults for optional address fields if the hook needs them, rather than hardcoding credentials. A hook can also cache its configured session for subsequent calls. section 9.2.1
Reading a connection in a Python task
A PythonOperator can receive a connection ID through op_kwargs. Its callable can retrieve the record with BaseHook.get_connection(conn_id), then use the returned host, port, login, and password for an HTTP request. In an HTTP-to-MinIO task, S3Hook(aws_conn_id="s3") uses a separate connection for storage. MinIO’s nondefault address is supplied as endpoint_url in that connection’s extras. section 12.1.2 section 13.3.1
Listing 13.6 Downloading data from the Citi Bike REST API to MinIO
import json
import requests
from airflow.hooks.base import BaseHook
from airflow.sdk import DAG
from airflow.timetables.interval import CronDataIntervalTimetable
from airflow.operators.standard.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from requests.auth import HTTPBasicAuth
def _download_citi_bike_data(data_interval_start, **_): #1
citibike_conn = BaseHook.get_connection(conn_id="citibike") #2
url = f"http://{citibike_conn.host}:{citibike_conn.port}/recent/minute/15"
response = requests.get(url, auth=HTTPBasicAuth(citibike_conn.login,
➥ citibike_conn.password))
data = response.json()
s3_hook = S3Hook(aws_conn_id="s3") #3
s3_hook.load_string(
string_data=json.dumps(data),
key=f"raw/citibike/{{data_interval_start.
➥ strftime("%Y%m%dT%H%M%S")} }.json", #4
bucket_name="datalake"
)
download_citi_bike_data = PythonOperator(
task_id="download_citi_bike_data",
python_callable=_download_citi_bike_data,
)
The download task shows direct BaseHook lookup for the API alongside an S3Hook configured by its own connection ID. section 13.3.1
The Airflow metastore is a convenient central place for credentials when the deployment is secured. Connection details can also be supplied through environment variables. For organizations using an external secrets manager, an Airflow secrets backend can make externally stored connections available to DAGs through the same connection interface; that backend needs its own configuration and storage conventions. section 12.1.2 section 16.5 section 13.3.1
