Python para DevOps: Docker, GitHub Actions y Automatización CI/CD

Python para DevOps: Docker, GitHub Actions y Automatización CI/CD

Python es el lenguaje preferido en DevOps gracias a su SDK de Docker, su amplio ecosistema de automatización y su integración nativa con las plataformas CI/CD más populares.

Docker SDK para Python

pip install docker

import docker

client = docker.from_env()   # conecta al daemon Docker local

# Listar contenedores en ejecución
for c in client.containers.list():
    print(f"{c.name}: {c.status} — {c.image.tags}")

# Listar imágenes
for img in client.images.list():
    print(f"{img.tags} — {img.short_id}")

# Ejecutar contenedor y capturar salida
output = client.containers.run(
    'python:3.12-slim',
    command='python -c "print(\'Hola desde Docker\')"',
    remove=True,           # borrar contenedor al terminar
    stdout=True,
)
print(output.decode())

# Ejecutar en background
contenedor = client.containers.run(
    'nginx:alpine',
    detach=True,
    ports={'80/tcp': 8080},   # host_port: container_port
    name='mi-nginx',
    environment={'NGINX_HOST': 'localhost'},
)
print(f"Contenedor iniciado: {contenedor.id[:12]}")
contenedor.stop()
contenedor.remove()

Construir imágenes

import docker
import io

# Dockerfile en memoria
dockerfile = '''
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "main.py"]
'''

client = docker.from_env()

# Desde un directorio
imagen, logs = client.images.build(
    path='.',
    tag='mi-app:latest',
    buildargs={'VERSION': '1.0'},
    rm=True,          # borrar contenedores intermedios
    nocache=False,
)
for log in logs:
    if 'stream' in log:
        print(log['stream'], end='')

print(f"Imagen construida: {imagen.id[:12]}")

# Publicar a registry
client.images.push('mi-usuario/mi-app', tag='latest')

Gestionar redes y volúmenes

import docker

client = docker.from_env()

# Crear red
red = client.networks.create('mi-red', driver='bridge')

# Crear volumen
volumen = client.volumes.create('datos-persistentes')

# Contenedor con red y volumen
contenedor = client.containers.run(
    'postgres:16-alpine',
    detach=True,
    name='mi-postgres',
    network='mi-red',
    volumes={'datos-persistentes': {'bind': '/var/lib/postgresql/data', 'mode': 'rw'}},
    environment={
        'POSTGRES_DB':       'mi_bd',
        'POSTGRES_USER':     'usuario',
        'POSTGRES_PASSWORD': 'secreto',
    },
    ports={'5432/tcp': 5432},
)

# Ejecutar comando en contenedor en ejecución
exit_code, output = contenedor.exec_run('pg_isready -U usuario')
print(f"PostgreSQL listo: {exit_code == 0} — {output.decode().strip()}")

# Limpiar
contenedor.stop()
contenedor.remove()
red.remove()
volumen.remove()

Automatización con subprocess — Docker CLI

import subprocess
import json
from typing import Any

def docker_run(imagen: str, comando: list[str], **kwargs) -> str:
    """Ejecuta un comando en un contenedor Docker y devuelve stdout."""
    cmd = ['docker', 'run', '--rm']
    for k, v in kwargs.items():
        cmd.extend([f'--{k}', v])
    cmd.append(imagen)
    cmd.extend(comando)
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return result.stdout.strip()

def docker_ps() -> list[dict]:
    """Lista contenedores en formato JSON."""
    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]

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

GitHub Actions — workflows en YAML

# .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:
      - name: Checkout código
        uses: actions/checkout@v4

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

      - name: Instalar dependencias
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-dev.txt

      - name: Linting con ruff
        run: ruff check .

      - name: Formateo con black
        run: black --check .

      - name: Type checking con mypy
        run: mypy src/

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

      - name: Subir cobertura a 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 a Docker Hub
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}

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

Script de deploy automatizado

#!/usr/bin/env python3
"""
deploy.py — Deploy automatizado a servidor remoto.
Uso: python deploy.py --env production --version 1.2.3
"""
import argparse
import subprocess
import sys
import logging
from pathlib import Path

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

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

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

    pasos = [
        # 1. Tests previos al deploy
        (['pytest', 'tests/', '-q', '--tb=short'], "Tests"),
        # 2. Build de la imagen
        (['docker', 'build', '-t', f'mi-app:{version}', '.'], "Build Docker"),
        # 3. Push al registry
        (['docker', 'push', f'registry.ejemplo.com/mi-app:{version}'], "Push registry"),
        # 4. Actualizar servicio en el servidor
        (['ssh', f'deploy@{env}.ejemplo.com',
          f'docker pull registry.ejemplo.com/mi-app:{version} && '
          f'docker service update --image registry.ejemplo.com/mi-app:{version} mi-app-service'],
         "Deploy remoto"),
    ]

    for cmd, nombre in pasos:
        log.info(f"=== {nombre} ===")
        if dry_run:
            log.info(f"[DRY RUN] {' '.join(cmd)}")
            continue
        try:
            run(cmd)
        except subprocess.CalledProcessError as e:
            log.error(f"Fallo en '{nombre}': código {e.returncode}")
            sys.exit(1)

    log.info(f"Deploy {version} completado en {env}")

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)

Gestión de configuración por entorno

# config.py — configuración tipada con pydantic-settings
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,
    )

    # Aplicación
    app_name: str = 'mi-app'
    debug:    bool = False
    log_level: str = 'INFO'

    # Base de datos
    database_url: str

    # Docker / infraestructura
    docker_registry: str = 'registry.ejemplo.com'
    max_containers:  int = 10

    # Secrets (nunca con valor por defecto)
    secret_key: str
    api_key:    str

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

# Uso
settings = get_settings()
print(f"BD: {settings.database_url}")
print(f"Debug: {settings.debug}")

Health checks y monitoreo

import httpx     # pip install httpx
import asyncio
from dataclasses import dataclass

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

async def check_service(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(servicios: dict[str, str]) -> list[ServiceStatus]:
    async with httpx.AsyncClient() as client:
        resultados = await asyncio.gather(*[
            check_service(client, nombre, url)
            for nombre, url in servicios.items()
        ])
    return list(resultados)

servicios = {
    'API':         'https://api.ejemplo.com/health',
    'Base de datos': 'https://db.ejemplo.com/ping',
    'Cache':       'https://cache.ejemplo.com/ping',
}

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

Buenas prácticas

• Docker SDK para Python cuando necesitas control programático (crear, inspeccionar, destruir contenedores desde código); subprocess para one-liners simples.
• Secrets en GitHub Actions: nunca hardcodees tokens; usa ${{ secrets.MI_SECRETO }} y configúralos en Settings → Secrets del repositorio.
• Cache de pip en CI: actions/setup-python con cache: 'pip' reduce el tiempo de instalación de 2 min a 15 s en pipelines repetidos.
• pytest --tb=short en CI para output conciso; añade -x para parar en el primer fallo y acelerar feedback.
• pydantic-settings para validar y tipar la configuración desde variables de entorno — falla rápido en arranque si falta algo crítico.
• Health checks asíncronos con httpx: comprueba todos los servicios en paralelo en lugar de secuencial para reducir tiempo de monitoreo.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

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