Connection management for cloud services in Apache Airflow - Mini Project: Build & Apply

Connection management for cloud services

📖 Scenario: You are setting up Apache Airflow to connect to a cloud storage service. To do this, you need to create and manage a connection in Airflow that stores the credentials and endpoint details securely. This project will guide you through creating a connection, configuring it, using it in a DAG, and finally printing the connection details.

🎯 Goal: Build an Airflow setup where you create a cloud service connection, configure it with necessary details, use it in a DAG to retrieve connection info, and print the connection details.

📋 What You'll Learn

Create an Airflow connection with exact ID and parameters

Add configuration variables for the connection

Write a DAG that uses the connection to get connection info

Print the connection details in the DAG task

💡 Why This Matters

🌍 Real World

Managing connections in Airflow is essential for securely accessing cloud services like storage, databases, or APIs without hardcoding credentials.

💼 Career

DevOps and Data Engineering roles often require setting up and managing Airflow connections to automate workflows that interact with cloud platforms.

1

Create an Airflow connection for cloud storage

Create an Airflow connection with the connection ID cloud_storage_conn. Set the connection type to Google Cloud, host to https://storage.googleapis.com, and login to user123. Use the password pass123.

Apache Airflow

# Create the Airflow connection with ID 'cloud_storage_conn'
# Your code here

Hint

Use the Connection class from airflow.models to create the connection object with the exact parameters.

2

Add configuration variables for the connection

Create a variable called project_id and set it to my-gcp-project. Also create a variable called location and set it to us-central1.

Apache Airflow

from airflow.models import Connection

cloud_storage_conn = Connection(
    conn_id='cloud_storage_conn',
    conn_type='google_cloud_platform',
    host='https://storage.googleapis.com',
    login='user123',
    password='pass123'
)

# Create variable 'project_id' with value 'my-gcp-project'
# Create variable 'location' with value 'us-central1'
# Your code here

Hint

Simply assign the exact string values to the variables project_id and location.

3

Write a DAG that uses the connection to get connection info

Write a DAG named cloud_storage_dag with a single PythonOperator task called print_conn_info. In the task's Python function, use Airflow's BaseHook.get_connection to get the connection with ID cloud_storage_conn. Extract the host, login, and password from the connection object and store them in variables named conn_host, conn_login, and conn_password respectively.

Apache Airflow

from airflow.models import Connection
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.hooks.base import BaseHook

cloud_storage_conn = Connection(
    conn_id='cloud_storage_conn',
    conn_type='google_cloud_platform',
    host='https://storage.googleapis.com',
    login='user123',
    password='pass123'
)

project_id = 'my-gcp-project'
location = 'us-central1'

# Define the DAG 'cloud_storage_dag' and the PythonOperator task 'print_conn_info'
# In the task function, get the connection using BaseHook.get_connection('cloud_storage_conn')
# Extract host, login, and password into variables 'conn_host', 'conn_login', 'conn_password'
# Your code here

Hint

Define a Python function that calls BaseHook.get_connection('cloud_storage_conn') and extracts the host, login, and password. Then create a DAG and a PythonOperator that runs this function.

4

Print the connection details in the DAG task

In the print_connection_info function, add print statements to display the connection host, login, and password. The output should be exactly:
Host: https://storage.googleapis.com
Login: user123
Password: pass123

Apache Airflow

from airflow.models import Connection
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.hooks.base import BaseHook

cloud_storage_conn = Connection(
    conn_id='cloud_storage_conn',
    conn_type='google_cloud_platform',
    host='https://storage.googleapis.com',
    login='user123',
    password='pass123'
)

project_id = 'my-gcp-project'
location = 'us-central1'

def print_connection_info():
    conn = BaseHook.get_connection('cloud_storage_conn')
    conn_host = conn.host
    conn_login = conn.login
    conn_password = conn.password
    # Print the connection details exactly as specified
    # Your code here

with DAG(
    dag_id='cloud_storage_dag',
    start_date=days_ago(1),
    schedule_interval=null
) as dag:
    print_conn_info = PythonOperator(
        task_id='print_conn_info',
        python_callable=print_connection_info
    )

Hint

Use print(f"Host: {conn_host}") and similar print statements for login and password inside the function.

Practice

1. What is the main purpose of managing connections in Airflow for cloud services?

easy

A. To manually enter credentials every time a task runs

B. To write custom code for each cloud service login

C. To store cloud login details securely and reuse them in tasks

D. To disable cloud service access in Airflow

Solution

1. Step 1: Understand connection management role

   Connection management stores credentials securely so tasks can access cloud services without exposing secrets.

2. Step 2: Identify reuse benefit

   Reusing connections avoids repeating credentials in multiple tasks, making workflows cleaner and safer.

3. Final Answer:

   To store cloud login details securely and reuse them in tasks -> Option C

4. Quick Check:

   Secure storage + reuse = A [OK]

Hint: Connections store credentials once for many tasks [OK]

Common Mistakes:

• Thinking credentials must be typed in each task
• Assuming connections are only for UI use
• Believing connections disable cloud access

2. Which command correctly adds a Google Cloud connection named 'gcp_default' in Airflow CLI?

easy

A. airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}'

B. airflow add connection gcp_default --type gcp --extra '{"key_path":"/path/key.json"}'

C. airflow connections create gcp_default --type google --extra '{"key":"/path/key.json"}'

D. airflow connection add gcp_default --conn-type gcp --conn-extra '{"key_path":"/path/key.json"}'

Solution

1. Step 1: Recall correct Airflow CLI syntax

   The command to add connections is 'airflow connections add' followed by connection id and parameters.

2. Step 2: Match correct flags and JSON format

   airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}' uses '--conn-type google_cloud_platform' and properly formatted JSON for extra parameters, matching Airflow docs.

3. Final Answer:

   airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}' -> Option A

4. Quick Check:

   Correct command syntax = C [OK]

Hint: Use 'airflow connections add' with correct flags [OK]

Common Mistakes:

• Using 'airflow add connection' instead of 'connections add'
• Wrong JSON key names in --conn-extra
• Missing quotes around JSON string

3. Given this Airflow DAG snippet using a connection ID 'aws_conn', what will happen if the connection is missing?

from airflow.providers.amazon.aws.operators.s3 import S3ListOperator

list_s3 = S3ListOperator(
    task_id='list_s3',
    bucket='my-bucket',
    aws_conn_id='aws_conn'
)

medium

A. The task will create a new connection automatically

B. The task will fail with a connection not found error

C. The task will skip without error

D. The task will run using default AWS credentials

Solution

1. Step 1: Understand connection usage in operator

   The operator uses 'aws_conn_id' to find credentials. If missing, it cannot authenticate.

2. Step 2: Predict behavior on missing connection

   Airflow raises an error when a required connection is not found, causing task failure.

3. Final Answer:

   The task will fail with a connection not found error -> Option B

4. Quick Check:

   Missing connection causes failure = D [OK]

Hint: Missing connection ID causes task failure [OK]

Common Mistakes:

• Assuming default credentials are used automatically
• Thinking task skips if connection missing
• Believing Airflow auto-creates connections

4. You added a connection with the command:

airflow connections add 'azure_default' --conn-type azure --conn-extra '{"tenantId":"abc","clientId":"123"}'

But your tasks fail to authenticate. What is the likely mistake?

medium

A. The JSON keys in --conn-extra are missing the 'extra__azure__' prefix

B. The connection type should be 'azure_cosmos' not 'azure'

C. The connection ID should not be quoted

D. The command should use 'airflow connection add' instead of 'connections add'

Solution

1. Step 1: Check connection type and extra format

   Azure connections require extra keys prefixed with 'extra__azure__' to be recognized properly.

2. Step 2: Identify JSON key format error

   Missing prefix in JSON keys causes Airflow to ignore credentials, leading to authentication failure.

3. Final Answer:

   The JSON keys in --conn-extra are missing the 'extra__azure__' prefix -> Option A

4. Quick Check:

   Correct extra key format fixes auth = B [OK]

Hint: Use 'extra__service__' prefix in --conn-extra keys [OK]

Common Mistakes:

• Using wrong connection type name
• Not prefixing JSON keys correctly
• Wrong CLI command syntax

5. You want to securely manage multiple cloud connections (AWS, GCP, Azure) in Airflow and reuse them in different DAGs. Which approach is best?

hard

A. Use environment variables only and avoid Airflow connections

B. Hardcode credentials in each DAG file for each cloud service

C. Store all credentials in a single connection with mixed cloud keys

D. Create separate connections for each cloud service using Airflow UI or CLI, then reference their IDs in tasks

Solution

1. Step 1: Understand best practice for connection management

   Separate connections per cloud service keep credentials organized and secure.

2. Step 2: Recognize reuse and security benefits

   Referencing connection IDs in tasks avoids duplication and keeps secrets out of code.

3. Final Answer:

   Create separate connections for each cloud service using Airflow UI or CLI, then reference their IDs in tasks -> Option D

4. Quick Check:

   Separate connections + reuse = A [OK]

Hint: Use separate connections per cloud, reference by ID [OK]

Common Mistakes:

• Hardcoding credentials in DAGs
• Mixing all credentials in one connection
• Avoiding Airflow connections for secrets