API et intégrations cloud avancées

Dans le chapitre précédent, on a utilisé les SDK cloud (Boto3, GCP, Azure, Docker) avec Python. Maintenant, on intègre Python avec Terraform, Ansible et Kubernetes pour un workflow complet.

🎯 Objectif : À la fin de ce chapitre, tu maîtriseras les concepts présentés ci-dessous.

Intégration avec Terraform

Python peut piloter Terraform pour du provisionnement d’infrastructure.

Exécuter Terraform depuis Python

import subprocess
import json
import os

class TerraformRunner:
    """Wrapper Python pour Terraform."""

    def __init__(self, working_dir):
        self.working_dir = working_dir

    def _run(self, command, **kwargs):
        """Exécute une commande Terraform."""
        cmd = ["terraform", "-chdir=" + self.working_dir] + command
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            **kwargs
        )
        if result.returncode != 0:
            raise RuntimeError(f"Terraform error: {result.stderr}")
        return result

    def init(self):
        """terraform init."""
        return self._run(["init", "-no-color"])

    def plan(self, var_file=None):
        """terraform plan."""
        cmd = ["plan", "-no-color", "-out=tfplan"]
        if var_file:
            cmd.append(f"-var-file={var_file}")
        return self._run(cmd)

    def apply(self, auto_approve=False):
        """terraform apply."""
        cmd = ["apply", "-no-color"]
        if auto_approve:
            cmd.append("-auto-approve")
        cmd.append("tfplan")
        return self._run(cmd)

    def destroy(self, auto_approve=False):
        """terraform destroy."""
        cmd = ["destroy", "-no-color"]
        if auto_approve:
            cmd.append("-auto-approve")
        return self._run(cmd)

    def output(self):
        """terraform output en JSON."""
        result = self._run(["output", "-json"])
        return json.loads(result.stdout)

    def state_list(self):
        """terraform state list."""
        result = self._run(["state", "list"])
        return result.stdout.strip().split("\n")


# Utilisation
tf = TerraformRunner("./infra/production")
tf.init()
tf.plan(var_file="production.tfvars")
tf.apply(auto_approve=True)

outputs = tf.output()
print(f"Load Balancer IP: {outputs['lb_ip']['value']}")
print(f"Database URL: {outputs['db_url']['value']}")

Générer du HCL depuis Python

import json

def generate_terraform_vars(env, config):
    """Génère un fichier .tfvars.json pour Terraform."""
    vars_data = {
        "environment": env,
        "region": config["region"],
        "instance_type": config.get("instance_type", "t3.micro"),
        "instance_count": config.get("instance_count", 1),
        "tags": {
            "Environment": env,
            "ManagedBy": "terraform",
            "Team": "devops"
        }
    }

    output_file = f"{env}.tfvars.json"
    with open(output_file, "w") as f:
        json.dump(vars_data, f, indent=2)

    print(f"✅ {output_file} généré")
    return output_file

# Utilisation
configs = {
    "staging": {"region": "eu-west-1", "instance_type": "t3.small", "instance_count": 2},
    "production": {"region": "eu-west-1", "instance_type": "t3.medium", "instance_count": 4},
}

for env, config in configs.items():
    generate_terraform_vars(env, config)

Intégration avec Ansible

Générer un inventaire dynamique

#!/usr/bin/env python3
"""
Inventaire Ansible dynamique — génère un inventaire depuis AWS EC2.
Usage: ansible-playbook -i inventory.py playbook.yml
"""

import json
import boto3

def get_inventory():
    """Génère un inventaire Ansible depuis les instances EC2."""
    ec2 = boto3.resource("ec2")
    inventory = {"_meta": {"hostvars": {}}}

    for instance in ec2.instances.filter(
        Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
    ):
        # Récupérer les tags
        tags = {t["Key"]: t["Value"] for t in (instance.tags or [])}
        name = tags.get("Name", instance.id)
        role = tags.get("Role", "ungrouped")
        env = tags.get("Environment", "unknown")

        # Créer les groupes
        for group in [role, env, f"{env}_{role}"]:
            if group not in inventory:
                inventory[group] = {"hosts": [], "vars": {}}
            inventory[group]["hosts"].append(name)

        # Variables d'hôte
        inventory["_meta"]["hostvars"][name] = {
            "ansible_host": instance.private_ip_address or instance.public_ip_address,
            "ansible_user": "ubuntu",
            "instance_id": instance.id,
            "instance_type": instance.instance_type,
            "availability_zone": instance.placement["AvailabilityZone"],
            "private_ip": instance.private_ip_address,
            "public_ip": instance.public_ip_address,
        }

    return inventory

if __name__ == "__main__":
    import sys
    if "--list" in sys.argv:
        print(json.dumps(get_inventory(), indent=2))
    elif "--host" in sys.argv:
        print(json.dumps({}))

Exécuter Ansible depuis Python

import subprocess
import json

def run_ansible_playbook(playbook, inventory, extra_vars=None, limit=None, tags=None):
    """Exécute un playbook Ansible."""
    cmd = ["ansible-playbook", playbook, "-i", inventory]

    if extra_vars:
        cmd.extend(["--extra-vars", json.dumps(extra_vars)])

    if limit:
        cmd.extend(["--limit", limit])

    if tags:
        cmd.extend(["--tags", ",".join(tags)])

    result = subprocess.run(cmd, capture_output=True, text=True)

    return {
        "success": result.returncode == 0,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "returncode": result.returncode
    }

# Utilisation
result = run_ansible_playbook(
    playbook="deploy.yml",
    inventory="inventory.py",
    extra_vars={"app_version": "v2.1.0", "env": "production"},
    tags=["deploy", "restart"]
)

if result["success"]:
    print("✅ Playbook exécuté avec succès")
else:
    print(f"❌ Erreur (code {result['returncode']})")
    print(result["stderr"])

Kubernetes avec Python

pip install kubernetes

from kubernetes import client, config

# Charger la config (kubeconfig ou in-cluster)
config.load_kube_config()  # Depuis ~/.kube/config
# config.load_incluster_config()  # Depuis un pod

v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()

# Lister les pods
pods = v1.list_namespaced_pod("production")
for pod in pods.items:
    print(f"  {pod.metadata.name:40} {pod.status.phase:10} {pod.status.pod_ip}")

# Lister les nodes
nodes = v1.list_node()
for node in nodes.items:
    conditions = {c.type: c.status for c in node.status.conditions}
    ready = conditions.get("Ready", "Unknown")
    print(f"  {node.metadata.name:30} Ready={ready}")

# Lister les déploiements
deployments = apps_v1.list_namespaced_deployment("production")
for dep in deployments.items:
    ready = dep.status.ready_replicas or 0
    desired = dep.spec.replicas
    print(f"  {dep.metadata.name:30} {ready}/{desired} replicas")

# Scaler un déploiement
body = {"spec": {"replicas": 5}}
apps_v1.patch_namespaced_deployment_scale(
    name="my-api",
    namespace="production",
    body=body
)
print("✅ Scaled to 5 replicas")

# Obtenir les logs d'un pod
logs = v1.read_namespaced_pod_log(
    name="my-api-7b9f5c6d4-abc12",
    namespace="production",
    tail_lines=50
)
print(logs)

# Créer un namespace
namespace = client.V1Namespace(
    metadata=client.V1ObjectMeta(name="staging")
)
v1.create_namespace(body=namespace)

Exemple complet : provisionneur multi-cloud

#!/usr/bin/env python3
"""
Provisionneur multi-cloud — crée des ressources sur AWS, GCP ou Azure.
"""

import json
from abc import ABC, abstractmethod


class CloudProvider(ABC):
    """Interface abstraite pour les providers cloud."""

    @abstractmethod
    def create_instance(self, name, instance_type, image):
        pass

    @abstractmethod
    def list_instances(self):
        pass

    @abstractmethod
    def create_bucket(self, name, region):
        pass


class AWSProvider(CloudProvider):
    def __init__(self, region="eu-west-1"):
        import boto3
        self.ec2 = boto3.resource("ec2", region_name=region)
        self.s3 = boto3.client("s3", region_name=region)
        self.region = region

    def create_instance(self, name, instance_type="t3.micro", image="ami-0abcdef"):
        instances = self.ec2.create_instances(
            ImageId=image,
            InstanceType=instance_type,
            MinCount=1, MaxCount=1,
            TagSpecifications=[{
                "ResourceType": "instance",
                "Tags": [{"Key": "Name", "Value": name}]
            }]
        )
        return {"id": instances[0].id, "provider": "aws"}

    def list_instances(self):
        return [
            {"id": i.id, "state": i.state["Name"], "type": i.instance_type}
            for i in self.ec2.instances.all()
        ]

    def create_bucket(self, name, region=None):
        self.s3.create_bucket(
            Bucket=name,
            CreateBucketConfiguration={"LocationConstraint": region or self.region}
        )
        return {"name": name, "provider": "aws"}


class GCPProvider(CloudProvider):
    def __init__(self, project, zone="europe-west6-a"):
        from google.cloud import compute_v1, storage
        self.project = project
        self.zone = zone
        self.instances_client = compute_v1.InstancesClient()
        self.storage_client = storage.Client()

    def create_instance(self, name, instance_type="e2-micro", image="debian-12"):
        # Simplifié — voir l'exemple complet plus haut
        print(f"Création de {name} sur GCP ({instance_type})")
        return {"name": name, "provider": "gcp"}

    def list_instances(self):
        instances = self.instances_client.list(project=self.project, zone=self.zone)
        return [{"name": i.name, "status": i.status} for i in instances]

    def create_bucket(self, name, region="europe-west6"):
        bucket = self.storage_client.create_bucket(name, location=region)
        return {"name": bucket.name, "provider": "gcp"}


def provision_infrastructure(provider, config):
    """Provisionne l'infrastructure selon un fichier de configuration."""
    print(f"🚀 Provisionnement avec {provider.__class__.__name__}")

    # Créer les instances
    for instance in config.get("instances", []):
        result = provider.create_instance(
            name=instance["name"],
            instance_type=instance.get("type", "t3.micro")
        )
        print(f"  ✅ Instance créée : {result}")

    # Créer les buckets
    for bucket in config.get("buckets", []):
        result = provider.create_bucket(bucket["name"], bucket.get("region"))
        print(f"  ✅ Bucket créé : {result}")


# Configuration
config = {
    "instances": [
        {"name": "web-01", "type": "t3.small"},
        {"name": "web-02", "type": "t3.small"},
        {"name": "db-01", "type": "t3.medium"}
    ],
    "buckets": [
        {"name": "app-assets-prod", "region": "eu-west-1"},
        {"name": "app-backups-prod", "region": "eu-west-1"}
    ]
}

# Utilisation
aws = AWSProvider(region="eu-west-1")
provision_infrastructure(aws, config)

À toi de jouer

Exercice 1 — Inventaire Ansible dynamique

Crée un script Python qui génère un inventaire Ansible JSON dynamique à partir d’une liste de serveurs dans un fichier YAML. Le script doit regrouper les serveurs par environnement (staging/prod) et ajouter les variables de groupe.

Exercice 2 — Wrapper Terraform

Écris un script Python qui exécute terraform plan avec subprocess, parse la sortie pour compter les ressources à créer/modifier/supprimer, et affiche un résumé coloré. Gère les erreurs Terraform proprement.

Exercice 3 — Défi bonus : client Kubernetes

Utilise la librairie kubernetes pour écrire un script qui liste tous les Pods en état CrashLoopBackOff dans un cluster, affiche les logs des derniers restarts, et génère un rapport Markdown.

Conclusion

Tu as maintenant les clés pour interagir avec l’infrastructure cloud depuis Python :

• Boto3 — S3, EC2, CloudWatch, IAM, SSM, SQS
• Google Cloud SDK — Storage, Compute, Logging
• Azure SDK — Blob Storage, VMs
• Docker SDK — Conteneurs, images, build
• Terraform — Wrapper Python, génération de configs
• Ansible — Inventaire dynamique, exécution de playbooks
• Kubernetes — API Python officielle

Chaque SDK suit le même pattern : authentification → client → opérations. Une fois que tu en maîtrises un, les autres sont faciles à prendre en main.

C’est la fin de la série Python pour DevOps. Tu as tout ce qu’il faut pour automatiser ton infrastructure, de la base du langage jusqu’aux SDK cloud. La prochaine étape ? Écrire tes propres outils et les mettre en production. 🚀

📝 Résumé

• boto3 (AWS), google-cloud-, azure- = les SDK pour piloter le cloud en Python
• Les SDK utilisent les credentials locaux (AWS CLI, gcloud, az login)
• Pagination obligatoire pour les listes longues (sinon tu ne récupères que la première page)
• Les waiters/pollers attendent qu’une ressource soit prête (instance running, bucket créé)
• Toujours gérer les exceptions spécifiques du SDK (botocore.exceptions.ClientError)

💡 À retenir : SDK cloud vs Terraform ? Le SDK est pour les scripts ponctuels et l’automatisation custom (nettoyage, audit, monitoring). Terraform est pour la gestion déclarative de l’infrastructure. Ils sont complémentaires, pas en compétition.

➡️ La suite

Tu maîtrises Python pour le DevOps ! On passe à FastAPI pour construire des API REST modernes — parce que le DevOps moderne, c’est aussi des microservices à développer et déployer.