GitOps Patterns with Python — Deep Dive

Implement GitOps workflows with Python — image tag updaters, Kubernetes operators with Kopf, drift detection, and config generation with CDK8s

Image tag updater

The most common Python component in a GitOps pipeline is the image updater. After CI builds and pushes a new container image, a Python script updates the image tag in the config repository, creating the Git commit that triggers deployment:

import subprocess
import tempfile
import logging
from pathlib import Path

logger = logging.getLogger(__name__)


class GitOpsImageUpdater:
    def __init__(
        self,
        config_repo: str,
        branch: str = "main",
        git_user: str = "gitops-bot",
        git_email: str = "gitops@example.com",
    ):
        self.config_repo = config_repo
        self.branch = branch
        self.git_user = git_user
        self.git_email = git_email
    
    def update_image_tag(
        self,
        app_name: str,
        new_tag: str,
        file_path: str = "apps/{app}/deployment.yaml",
        image_pattern: str = "image: {registry}/{app}:",
    ) -> dict:
        """Clone config repo, update image tag, commit and push."""
        with tempfile.TemporaryDirectory() as tmpdir:
            # Clone
            self._run(
                f"git clone --depth 1 -b {self.branch} {self.config_repo} repo",
                cwd=tmpdir,
            )
            repo_dir = Path(tmpdir) / "repo"
            
            # Configure git
            self._run(f"git config user.name '{self.git_user}'", cwd=repo_dir)
            self._run(f"git config user.email '{self.git_email}'", cwd=repo_dir)
            
            # Find and update the manifest
            manifest = repo_dir / file_path.format(app=app_name)
            if not manifest.exists():
                return {"updated": False, "reason": f"{manifest} not found"}
            
            content = manifest.read_text()
            
            # Replace image tag using line-by-line matching
            lines = content.split("\n")
            updated = False
            registry_prefix = image_pattern.format(
                registry="", app=app_name
            ).rstrip(":")
            
            for i, line in enumerate(lines):
                stripped = line.strip()
                if stripped.startswith("image:") and app_name in stripped:
                    # Extract everything before the tag
                    prefix = stripped.rsplit(":", 1)[0]
                    indent = line[: len(line) - len(line.lstrip())]
                    lines[i] = f"{indent}{prefix}:{new_tag}"
                    updated = True
                    break
            
            if not updated:
                return {"updated": False, "reason": "image line not found"}
            
            manifest.write_text("\n".join(lines))
            
            # Commit and push
            self._run(f"git add {manifest.name}", cwd=manifest.parent)
            commit_msg = f"chore({app_name}): update image tag to {new_tag}"
            self._run(f"git commit -m '{commit_msg}'", cwd=repo_dir)
            self._run(f"git push origin {self.branch}", cwd=repo_dir)
            
            return {
                "updated": True,
                "app": app_name,
                "tag": new_tag,
                "commit_message": commit_msg,
            }
    
    def _run(self, cmd: str, cwd=None) -> subprocess.CompletedProcess:
        return subprocess.run(
            cmd, shell=True, cwd=cwd, check=True,
            capture_output=True, text=True,
        )

Kubernetes operator with Kopf

For custom resources that ArgoCD and Flux can’t natively reconcile, Python operators fill the gap. Kopf (Kubernetes Operator Pythonic Framework) makes this accessible:

pip install kopf kubernetes
# operator.py — GitOps-aware custom resource operator
import kopf
import kubernetes
import logging
import subprocess
import tempfile
from pathlib import Path

logger = logging.getLogger(__name__)


@kopf.on.create("example.com", "v1", "gitopsapps")
@kopf.on.update("example.com", "v1", "gitopsapps")
async def reconcile_gitops_app(spec, name, namespace, **kwargs):
    """Reconcile a GitOpsApp custom resource.
    
    The CRD spec contains:
      - repoURL: git repository URL
      - path: path within repo to manifests
      - targetRevision: branch or tag
      - syncPolicy: auto or manual
    """
    repo_url = spec.get("repoURL")
    path = spec.get("path", ".")
    revision = spec.get("targetRevision", "main")
    
    logger.info(f"Reconciling {name}: {repo_url}@{revision}/{path}")
    
    with tempfile.TemporaryDirectory() as tmpdir:
        # Clone the config repo
        subprocess.run(
            ["git", "clone", "--depth", "1", "-b", revision, repo_url, "repo"],
            cwd=tmpdir,
            check=True,
            capture_output=True,
        )
        
        manifest_dir = Path(tmpdir) / "repo" / path
        
        if not manifest_dir.exists():
            raise kopf.PermanentError(f"Path {path} not found in repo")
        
        # Apply all YAML files in the path
        manifests = list(manifest_dir.glob("*.yaml")) + list(
            manifest_dir.glob("*.yml")
        )
        
        applied = []
        for manifest in sorted(manifests):
            result = subprocess.run(
                ["kubectl", "apply", "-f", str(manifest), "-n", namespace],
                capture_output=True,
                text=True,
            )
            if result.returncode == 0:
                applied.append(manifest.name)
            else:
                logger.error(f"Failed to apply {manifest.name}: {result.stderr}")
    
    return {"applied": applied, "revision": revision}


@kopf.timer("example.com", "v1", "gitopsapps", interval=300)
async def drift_check(spec, name, namespace, **kwargs):
    """Periodically check for drift between Git and actual state."""
    logger.info(f"Drift check for {name}")
    # Re-run reconciliation — Kubernetes apply is idempotent
    await reconcile_gitops_app(spec, name, namespace, **kwargs)

Config generation with CDK8s

CDK8s lets you define Kubernetes resources in Python with full type safety and abstraction:

pip install cdk8s cdk8s-plus-27
# main.py — Generate Kubernetes manifests from Python
from cdk8s import App, Chart
from cdk8s_plus_27 import (
    Deployment,
    Service,
    ServiceType,
    ContainerProps,
    Probe,
    ResourceRequirements,
    CpuAmount,
    MemoryAmount,
    HorizontalPodAutoscaler,
)
from constructs import Construct


class WebService(Chart):
    def __init__(
        self,
        scope: Construct,
        id: str,
        *,
        image: str,
        replicas: int = 3,
        port: int = 8000,
        cpu_request: str = "100m",
        memory_request: str = "128Mi",
    ):
        super().__init__(scope, id)
        
        deployment = Deployment(
            self,
            "deployment",
            replicas=replicas,
            containers=[
                ContainerProps(
                    name="app",
                    image=image,
                    port_number=port,
                    readiness=Probe.from_http_get(
                        path="/health",
                        port=port,
                    ),
                    liveness=Probe.from_http_get(
                        path="/health",
                        port=port,
                    ),
                    resources=ResourceRequirements(
                        cpu=CpuAmount.milli(int(cpu_request.rstrip("m"))),
                        memory=MemoryAmount.mebibytes(
                            int(memory_request.rstrip("Mi"))
                        ),
                    ),
                )
            ],
        )
        
        service = Service(
            self,
            "service",
            selector=deployment,
            service_type=ServiceType.CLUSTER_IP,
            ports=[{"port": 80, "target_port": port}],
        )
        
        HorizontalPodAutoscaler(
            self,
            "hpa",
            target=deployment,
            min_replicas=replicas,
            max_replicas=replicas * 3,
            metrics=[
                {"type": "cpu", "target": {"type": "utilization", "average_utilization": 70}},
            ],
        )


# Generate manifests for different environments
app = App()

WebService(
    app,
    "payment-service-prod",
    image="registry.example.com/payment:v1.2.3",
    replicas=3,
    cpu_request="500m",
    memory_request="512Mi",
)

WebService(
    app,
    "payment-service-staging",
    image="registry.example.com/payment:v1.3.0-rc1",
    replicas=1,
    cpu_request="100m",
    memory_request="128Mi",
)

app.synth()  # Generates YAML files in dist/

Commit the generated dist/ directory to the config repo. ArgoCD or Flux picks up the changes.

Drift detection and alerting

A standalone drift detector that compares Git state with cluster state:

from kubernetes import client, config
import subprocess
import tempfile
import json
import logging
from pathlib import Path
from deepdiff import DeepDiff

logger = logging.getLogger(__name__)


class DriftDetector:
    def __init__(self, config_repo: str, branch: str = "main"):
        self.config_repo = config_repo
        self.branch = branch
        config.load_incluster_config()
        self.apps = client.AppsV1Api()
    
    def detect_drift(self, namespace: str, app_path: str) -> list[dict]:
        """Compare desired state in Git with actual state in cluster."""
        drifts = []
        
        with tempfile.TemporaryDirectory() as tmpdir:
            subprocess.run(
                ["git", "clone", "--depth", "1", "-b", self.branch,
                 self.config_repo, "repo"],
                cwd=tmpdir, check=True, capture_output=True,
            )
            
            manifest_dir = Path(tmpdir) / "repo" / app_path
            
            for manifest_file in manifest_dir.glob("*.yaml"):
                import yaml
                
                with open(manifest_file) as f:
                    desired = yaml.safe_load(f)
                
                if not desired or desired.get("kind") != "Deployment":
                    continue
                
                name = desired["metadata"]["name"]
                
                try:
                    actual = self.apps.read_namespaced_deployment(
                        name, namespace
                    )
                    actual_dict = actual.to_dict()
                    
                    # Compare key fields
                    desired_image = (
                        desired["spec"]["template"]["spec"]["containers"][0]["image"]
                    )
                    actual_image = actual_dict["spec"]["template"]["spec"][
                        "containers"
                    ][0]["image"]
                    
                    desired_replicas = desired["spec"].get("replicas", 1)
                    actual_replicas = actual_dict["spec"].get("replicas", 1)
                    
                    if desired_image != actual_image:
                        drifts.append({
                            "resource": name,
                            "field": "image",
                            "desired": desired_image,
                            "actual": actual_image,
                        })
                    
                    if desired_replicas != actual_replicas:
                        drifts.append({
                            "resource": name,
                            "field": "replicas",
                            "desired": desired_replicas,
                            "actual": actual_replicas,
                        })
                    
                except client.ApiException as e:
                    if e.status == 404:
                        drifts.append({
                            "resource": name,
                            "field": "existence",
                            "desired": "present",
                            "actual": "missing",
                        })
        
        return drifts

GitOps workflow comparison

PatternToolPython roleBest for
Image updaterCustom scriptCore — updates Git commitsAny GitOps setup
Config generationCDK8s, KapitanCore — generates manifestsComplex multi-env configs
Custom operatorsKopfCore — reconciles CRDsNon-K8s resources
Drift detectionCustom scriptCore — compares stateCompliance, auditing
PR-based promotionGitHub Actions + PythonGlue — creates PRs for env promotionMulti-stage environments

Security considerations

GitOps shifts the security boundary. Since Git is the source of truth, protecting the config repository becomes critical:

  • Branch protection — require PR reviews for config changes
  • Signed commits — verify that commits come from authorized users
  • Sealed Secrets — encrypt secrets before storing in Git (using Bitnami Sealed Secrets or SOPS)
  • RBAC — the GitOps agent should have minimal cluster permissions
  • Audit logging — Git’s built-in history provides an immutable audit trail

Python scripts that update the config repo should use deploy keys with write access to only that repository, not broad personal access tokens.

The one thing to remember: Python is the glue language of GitOps — it generates configs, updates image tags, builds custom operators, and detects drift. The pattern works because Git provides an immutable audit trail and Kubernetes provides a declarative API that makes reconciliation natural.

pythongitopskubernetesdevops

See Also

Related Topics

ab-testing How tech companies run thousands of experiments at once to improve their products — the scientific method applied to everything from button colors to recommendation algorithms. 3 levels → activation-functions Why neural networks need these tiny mathematical functions — and how ReLU's simplicity accidentally made deep learning possible. 3 levels → ai-agents-architecture How AI systems go from answering questions to actually doing things — the design patterns that turn language models into autonomous agents that browse, code, and plan. 3 levels →