Metadata-Version: 2.3
Name: cstg-airflow-container-utils
Version: 0.2.0
Summary: Conditional usage of DockerOperator/KubernetesPodOpetor
Author: Felix Yman
Author-email: Felix Yman <felix.yman@capiostgoran.se>
Requires-Dist: apache-airflow>=3.0.2
Requires-Dist: apache-airflow-providers-cncf-kubernetes>=10.12.4
Requires-Dist: apache-airflow-providers-docker>=4.5.2
Requires-Python: >=3.12
Description-Content-Type: text/markdown

# `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

| Variable | Default | Description |
| :--- | :--- | :--- |
| `AIRFLOW_ENV` | `local` | Controls the execution mode. If set to `dev`, the function returns a `DockerOperator`. For any other value (including `local` or `prod`), it defaults to returning a `KubernetesPodOperator`. |

## 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

```python
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 
    )