get_container_operator

A factory function for Apache Airflow that abstracts containerized task execution. It dynamically returns either a DockerOperator (for local development) or a KubernetesPodOperator (for production/cluster environments) based on the active environment variable.

Description

When developing Airflow DAGs locally, it is often easier and faster to run tasks in local Docker containers rather than provisioning a full Kubernetes cluster. This function bridges the gap between local development and production deployments by providing a unified interface.

It reads the AIRFLOW_ENV environment variable to determine the execution context and provisions the appropriate operator with sensible defaults.

Parameters

• task_id (str): The unique identifier for the Airflow task.
• image (str): The Docker image to execute (e.g., my-repo/my-image:latest).
• environment (dict): A dictionary of environment variables to inject into the container at runtime.
• command (str): The command to execute inside the container.
  • Note: For Kubernetes deployments, this string is safely parsed into a list using shlex.split() to comply with the KubernetesPodOperator requirements.
• **kwargs: Additional keyword arguments to pass down to the underlying DockerOperator or KubernetesPodOperator.

Returns

• BaseOperator: Returns either a DockerOperator or a KubernetesPodOperator instance configured for the respective environment.

Environment Variables

Operator Configurations

Development Mode (AIRFLOW_ENV=dev)

Returns a DockerOperator configured to run on the local Docker daemon:

• Connects to: unix://var/run/docker.sock
• Network: airflow-docker_default
• Cleans up the container automatically after success (auto_remove="success").

Production Mode (Default/Fallback)

Returns a KubernetesPodOperator configured for in-cluster execution:

• Namespace: airflow
• Service Account: airflow-worker
• Image Pull Secrets: cstg-pullsecret
• Automatically deletes the pod after execution (is_delete_operator_pod=True).
• Fetches logs automatically (get_logs=True).

Usage Example

from datetime import datetime
from airflow import DAG
from cstg_airflow_container_utils.operators import get_container_operator

with DAG(
    dag_id="example_container_dag",
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:

    run_data_job = get_container_operator(
        task_id="run_data_processing",
        image="[my-registry.com/data-processor:v1.2](https://my-registry.com/data-processor:v1.2)",
        environment={"LOG_LEVEL": "INFO", "API_KEY": "secret"},
        command="python main.py --process-all",
        # kwargs passed to the underlying operator:
        retries=2 
    )