Python for DevOps: Docker, GitHub Actions and CI/CD Automation

Python for DevOps: Docker, GitHub Actions and CI/CD Automation

Python is DevOps's language of choice thanks to its Docker SDK, rich automation ecosystem, and native integration with the most popular CI/CD platforms.

Docker SDK for Python

pip install docker

import docker

client = docker.from_env()   # connects to the local Docker daemon

# List running containers
for c in client.containers.list():
    print(f"{c.name}: {c.status} — {c.image.tags}")

# Run a container and capture output
output = client.containers.run(
    'python:3.12-slim',
    command='python -c "print(\'Hello from Docker\')"',
    remove=True,       # delete container when done
    stdout=True,
)
print(output.decode())

# Run in the background
container = client.containers.run(
    'nginx:alpine',
    detach=True,
    ports={'80/tcp': 8080},
    name='my-nginx',
    environment={'NGINX_HOST': 'localhost'},
)
print(f"Container started: {container.id[:12]}")
container.stop()
container.remove()

Build images programmatically

import docker

client = docker.from_env()

image, logs = client.images.build(
    path='.',
    tag='my-app:latest',
    buildargs={'VERSION': '1.0'},
    rm=True,       # remove intermediate containers
)
for log in logs:
    if 'stream' in log:
        print(log['stream'], end='')

# Push to registry
client.images.push('myuser/my-app', tag='latest')

Manage networks and volumes

import docker

client = docker.from_env()

net  = client.networks.create('my-net', driver='bridge')
vol  = client.volumes.create('persistent-data')

container = client.containers.run(
    'postgres:16-alpine',
    detach=True,
    name='my-postgres',
    network='my-net',
    volumes={'persistent-data': {'bind': '/var/lib/postgresql/data', 'mode': 'rw'}},
    environment={
        'POSTGRES_DB':       'mydb',
        'POSTGRES_USER':     'user',
        'POSTGRES_PASSWORD': 'secret',
    },
    ports={'5432/tcp': 5432},
)

exit_code, output = container.exec_run('pg_isready -U user')
print(f"PostgreSQL ready: {exit_code == 0} — {output.decode().strip()}")

container.stop()
container.remove()
net.remove()
vol.remove()

Docker CLI via subprocess

import subprocess
import json

def docker_ps() -> list[dict]:
    result = subprocess.run(
        ['docker', 'ps', '--format', '{{json .}}'],
        capture_output=True, text=True, check=True
    )
    return [json.loads(line) for line in result.stdout.strip().splitlines() if line]

def docker_run(image: str, command: list[str], **flags) -> str:
    cmd = ['docker', 'run', '--rm']
    for k, v in flags.items():
        cmd.extend([f'--{k}', v])
    cmd.append(image)
    cmd.extend(command)
    return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout.strip()

for c in docker_ps():
    print(f"{c.get('Names')}: {c.get('Status')}")

GitHub Actions — CI/CD workflow

# .github/workflows/ci.yml
name: CI Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  PYTHON_VERSION: "3.12"

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: 'pip'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-dev.txt

      - name: Lint with ruff
        run: ruff check .

      - name: Format check with black
        run: black --check .

      - name: Type check with mypy
        run: mypy src/

      - name: Run pytest
        run: |
          pytest tests/ -v --tb=short \
            --cov=src \
            --cov-report=xml \
            --cov-report=term-missing

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v4
        with:
          files: coverage.xml

  build-docker:
    needs: lint-and-test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4

      - name: Login to Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}

      - name: Build and push image
        uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: |
            myuser/my-app:latest
            myuser/my-app:${{ github.sha }}
          cache-from: type=registry,ref=myuser/my-app:buildcache
          cache-to: type=registry,ref=myuser/my-app:buildcache,mode=max

Automated deploy script

#!/usr/bin/env python3
"""
deploy.py — Automated deploy to a remote server.
Usage: python deploy.py --env production --version 1.2.3
"""
import argparse
import subprocess
import sys
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
log = logging.getLogger('deploy')

def run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
    log.info(f"$ {' '.join(cmd)}")
    return subprocess.run(cmd, check=True, **kwargs)

def deploy(env: str, version: str, dry_run: bool = False):
    log.info(f"Deploying {version} to {env}")

    steps = [
        (['pytest', 'tests/', '-q', '--tb=short'],                      "Pre-deploy tests"),
        (['docker', 'build', '-t', f'my-app:{version}', '.'],           "Docker build"),
        (['docker', 'push', f'registry.example.com/my-app:{version}'],  "Registry push"),
        (['ssh', f'deploy@{env}.example.com',
          f'docker pull registry.example.com/my-app:{version} && '
          f'docker service update --image registry.example.com/my-app:{version} my-app-svc'],
         "Remote deploy"),
    ]

    for cmd, name in steps:
        log.info(f"=== {name} ===")
        if dry_run:
            log.info(f"[DRY RUN] {' '.join(cmd)}")
            continue
        try:
            run(cmd)
        except subprocess.CalledProcessError as e:
            log.error(f"Step '{name}' failed with code {e.returncode}")
            sys.exit(1)

    log.info(f"Deployment {version} to {env} complete")

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--env',     required=True, choices=['staging', 'production'])
    parser.add_argument('--version', required=True)
    parser.add_argument('--dry-run', action='store_true')
    args = parser.parse_args()
    deploy(args.env, args.version, args.dry_run)

Typed configuration management

# config.py — pydantic-settings for environment-based config
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        case_sensitive=False,
    )

    # App
    app_name:  str  = 'my-app'
    debug:     bool = False
    log_level: str  = 'INFO'

    # Database
    database_url: str

    # Secrets (no defaults — fail fast if missing)
    secret_key: str
    api_key:    str

@lru_cache
def get_settings() -> Settings:
    return Settings()

settings = get_settings()
print(f"DB: {settings.database_url}")

Async health checks

import httpx
import asyncio
from dataclasses import dataclass

@dataclass
class ServiceStatus:
    name:       str
    url:        str
    ok:         bool
    latency_ms: float
    error:      str = ''

async def check(client: httpx.AsyncClient, name: str, url: str) -> ServiceStatus:
    import time
    t0 = time.perf_counter()
    try:
        resp    = await client.get(url, timeout=5.0)
        latency = (time.perf_counter() - t0) * 1000
        return ServiceStatus(name, url, ok=resp.status_code < 400, latency_ms=latency)
    except Exception as e:
        latency = (time.perf_counter() - t0) * 1000
        return ServiceStatus(name, url, ok=False, latency_ms=latency, error=str(e))

async def health_check_all(services: dict[str, str]) -> list[ServiceStatus]:
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[
            check(client, name, url) for name, url in services.items()
        ])
    return list(results)

services = {
    'API':      'https://api.example.com/health',
    'Database': 'https://db.example.com/ping',
    'Cache':    'https://cache.example.com/ping',
}

for s in asyncio.run(health_check_all(services)):
    status = "OK" if s.ok else "FAIL"
    print(f"[{status}] {s.name} — {s.latency_ms:.0f}ms {s.error or ''}")

Best practices

• Docker SDK for Python when you need programmatic control; subprocess for simple one-off CLI calls.
• Secrets in GitHub Actions: never hardcode tokens — use ${{ secrets.MY_SECRET }} and configure them in repo Settings → Secrets.
• Pip cache in CI: actions/setup-python with cache: 'pip' cuts install time from 2 min to ~15 s on repeated runs.
• pytest --tb=short in CI for concise output; add -x to stop on first failure for faster feedback.
• pydantic-settings validates and types all config from environment variables — fails fast on startup if anything critical is missing.
• Async health checks with httpx: check all services in parallel instead of sequentially to minimize monitoring latency.

Related conversions

Frequent conversions across the catalogue:

• JPG to PNG
• MP4 to MP3
• PDF to DOCX
• PNG to JPG
• MP3 to WAV
• DOCX to PDF