SDK Cloud : AWS et GCP avec Python

🎯 Objectif : À la fin de ce chapitre, tu sauras utiliser les SDK cloud Python pour automatiser la gestion de ton infrastructure AWS, GCP et Azure. ⏱️ Durée estimée : 55 minutes | Niveau : Avancé

Introduction

Les cloud providers (AWS, GCP, Azure) exposent tous des SDK Python. Ces SDK te permettent de gérer ton infrastructure programmatiquement : créer des VMs, gérer des buckets S3, interagir avec Kubernetes, provisionner des ressources — tout ça depuis un script Python.

Ce chapitre est le pont entre Python et l’infrastructure. On couvre les SDK majeurs avec des exemples concrets et réutilisables.

Boto3 — AWS SDK pour Python

Boto3 est le SDK officiel d’Amazon Web Services. Il couvre l’intégralité des services AWS.

pip install boto3

Configuration

import boto3

# Option 1 : Credentials via fichier ~/.aws/credentials
# [default]
# aws_access_key_id = AKIA...
# aws_secret_access_key = abc123...
# region = eu-west-1

# Option 2 : Variables d'environnement
# export AWS_ACCESS_KEY_ID=AKIA...
# export AWS_SECRET_ACCESS_KEY=abc123...
# export AWS_DEFAULT_REGION=eu-west-1

# Option 3 : Explicite dans le code (déconseillé en prod)
session = boto3.Session(
    aws_access_key_id="AKIA...",
    aws_secret_access_key="abc123...",
    region_name="eu-west-1"
)

S3 — Gestion de buckets et objets

import boto3
from botocore.exceptions import ClientError

s3 = boto3.client("s3")

# Lister les buckets
response = s3.list_buckets()
for bucket in response["Buckets"]:
    print(f"  📦 {bucket['Name']} (créé le {bucket['CreationDate']})")

# Créer un bucket
s3.create_bucket(
    Bucket="my-devops-backups-2026",
    CreateBucketConfiguration={"LocationConstraint": "eu-west-1"}
)

# Uploader un fichier
s3.upload_file("backup.tar.gz", "my-devops-backups-2026", "backups/2026-03-20.tar.gz")

# Uploader avec des métadonnées
s3.upload_file(
    "config.yaml",
    "my-devops-backups-2026",
    "configs/config.yaml",
    ExtraArgs={
        "ServerSideEncryption": "AES256",
        "Metadata": {"environment": "production"}
    }
)

# Télécharger un fichier
s3.download_file("my-devops-backups-2026", "backups/2026-03-20.tar.gz", "/tmp/backup.tar.gz")

# Lister les objets d'un bucket
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="my-devops-backups-2026", Prefix="backups/"):
    for obj in page.get("Contents", []):
        size_mb = obj["Size"] / 1024 / 1024
        print(f"  {obj['Key']} — {size_mb:.2f} MB — {obj['LastModified']}")

# Supprimer un objet
s3.delete_object(Bucket="my-devops-backups-2026", Key="backups/old-backup.tar.gz")

# Générer une URL pré-signée (accès temporaire)
url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-devops-backups-2026", "Key": "backups/2026-03-20.tar.gz"},
    ExpiresIn=3600  # 1 heure
)
print(f"URL temporaire : {url}")

EC2 — Gestion des instances

ec2 = boto3.resource("ec2")
ec2_client = boto3.client("ec2")

# Lister les instances
for instance in ec2.instances.all():
    name = ""
    if instance.tags:
        name = next((t["Value"] for t in instance.tags if t["Key"] == "Name"), "")
    print(f"  {instance.id} | {name:20} | {instance.state['Name']:10} | {instance.instance_type}")

# Lancer une instance
instances = ec2.create_instances(
    ImageId="ami-0abcdef1234567890",
    InstanceType="t3.micro",
    MinCount=1,
    MaxCount=1,
    KeyName="my-key",
    SecurityGroupIds=["sg-12345678"],
    SubnetId="subnet-12345678",
    TagSpecifications=[{
        "ResourceType": "instance",
        "Tags": [
            {"Key": "Name", "Value": "web-server-01"},
            {"Key": "Environment", "Value": "production"},
            {"Key": "ManagedBy", "Value": "python-sdk"}
        ]
    }]
)
instance = instances[0]
print(f"Instance lancée : {instance.id}")

# Attendre que l'instance soit running
instance.wait_until_running()
instance.reload()
print(f"IP publique : {instance.public_ip_address}")

# Arrêter / démarrer / terminer
instance.stop()
instance.start()
# instance.terminate()  # ⚠️ Irréversible

Autres services AWS courants

# CloudWatch — Métriques et alarmes
cloudwatch = boto3.client("cloudwatch")

# Récupérer les métriques CPU d'une instance
from datetime import datetime, timedelta

response = cloudwatch.get_metric_statistics(
    Namespace="AWS/EC2",
    MetricName="CPUUtilization",
    Dimensions=[{"Name": "InstanceId", "Value": "i-1234567890abcdef0"}],
    StartTime=datetime.utcnow() - timedelta(hours=1),
    EndTime=datetime.utcnow(),
    Period=300,
    Statistics=["Average"]
)

for point in sorted(response["Datapoints"], key=lambda x: x["Timestamp"]):
    print(f"  {point['Timestamp']}: {point['Average']:.1f}%")


# IAM — Gestion des utilisateurs
iam = boto3.client("iam")
users = iam.list_users()
for user in users["Users"]:
    print(f"  {user['UserName']} — créé le {user['CreateDate']}")


# SSM — Parameter Store
ssm = boto3.client("ssm")

# Stocker un secret
ssm.put_parameter(
    Name="/myapp/production/db_password",
    Value="super-secret-password",
    Type="SecureString",
    Overwrite=True
)

# Récupérer un secret
param = ssm.get_parameter(Name="/myapp/production/db_password", WithDecryption=True)
print(param["Parameter"]["Value"])


# SQS — Files d'attente
sqs = boto3.client("sqs")
queue_url = "https://sqs.eu-west-1.amazonaws.com/123456789/my-queue"

# Envoyer un message
sqs.send_message(QueueUrl=queue_url, MessageBody='{"action": "deploy", "service": "api"}')

# Recevoir un message
messages = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1)
for msg in messages.get("Messages", []):
    print(f"Message : {msg['Body']}")
    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])

Google Cloud SDK pour Python

pip install google-cloud-storage google-cloud-compute google-cloud-logging

Authentication

# Option 1 : Service Account (recommandé en prod)
# export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"

# Option 2 : gcloud CLI
# gcloud auth application-default login

Cloud Storage

from google.cloud import storage

client = storage.Client()

# Lister les buckets
for bucket in client.list_buckets():
    print(f"  📦 {bucket.name}")

# Créer un bucket
bucket = client.create_bucket("my-devops-backups-ch", location="europe-west6")  # Zurich

# Uploader un fichier
bucket = client.bucket("my-devops-backups-ch")
blob = bucket.blob("backups/2026-03-20.tar.gz")
blob.upload_from_filename("backup.tar.gz")

# Télécharger un fichier
blob = bucket.blob("backups/2026-03-20.tar.gz")
blob.download_to_filename("/tmp/backup.tar.gz")

# Lister les objets
blobs = bucket.list_blobs(prefix="backups/")
for blob in blobs:
    print(f"  {blob.name} — {blob.size / 1024 / 1024:.2f} MB")

# URL signée
from datetime import timedelta
url = blob.generate_signed_url(expiration=timedelta(hours=1))

Compute Engine

from google.cloud import compute_v1

instances_client = compute_v1.InstancesClient()

# Lister les instances
project = "my-project-id"
zone = "europe-west6-a"

instances = instances_client.list(project=project, zone=zone)
for instance in instances:
    print(f"  {instance.name} | {instance.status} | {instance.machine_type.split('/')[-1]}")

# Créer une instance
def create_instance(project, zone, name, machine_type="e2-micro"):
    """Crée une instance Compute Engine."""
    instance = compute_v1.Instance()
    instance.name = name
    instance.machine_type = f"zones/{zone}/machineTypes/{machine_type}"

    # Disque de boot
    disk = compute_v1.AttachedDisk()
    disk.auto_delete = True
    disk.boot = True
    init_params = compute_v1.AttachedDiskInitializeParams()
    init_params.source_image = "projects/debian-cloud/global/images/family/debian-12"
    init_params.disk_size_gb = 20
    disk.initialize_params = init_params
    instance.disks = [disk]

    # Réseau
    network_interface = compute_v1.NetworkInterface()
    network_interface.name = "global/networks/default"
    access_config = compute_v1.AccessConfig()
    access_config.name = "External NAT"
    access_config.type_ = "ONE_TO_ONE_NAT"
    network_interface.access_configs = [access_config]
    instance.network_interfaces = [network_interface]

    # Labels
    instance.labels = {"env": "dev", "managed-by": "python-sdk"}

    operation = instances_client.insert(project=project, zone=zone, instance_resource=instance)
    operation.result()  # Attendre la fin
    print(f"✅ Instance {name} créée")

create_instance(project, zone, "web-server-01")

Cloud Logging

from google.cloud import logging as cloud_logging

client = cloud_logging.Client()

# Lire les logs
for entry in client.list_entries(
    filter_='resource.type="gce_instance" AND severity>=WARNING',
    max_results=10
):
    print(f"  [{entry.severity}] {entry.timestamp}: {entry.payload}")

# Écrire un log
logger = client.logger("my-app")
logger.log_text("Déploiement terminé avec succès", severity="INFO")
logger.log_struct({
    "action": "deploy",
    "service": "api",
    "version": "v2.1.0",
    "status": "success"
}, severity="INFO")

Azure SDK pour Python

pip install azure-identity azure-mgmt-compute azure-mgmt-resource azure-storage-blob

Authentication

from azure.identity import DefaultAzureCredential

# Utilise automatiquement :
# 1. Variables d'environnement (AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET)
# 2. Managed Identity (en prod sur Azure)
# 3. Azure CLI (az login)
credential = DefaultAzureCredential()

Blob Storage

from azure.storage.blob import BlobServiceClient

connection_string = "DefaultEndpointsProtocol=https;AccountName=myaccount;..."
blob_service = BlobServiceClient.from_connection_string(connection_string)

# Lister les conteneurs
for container in blob_service.list_containers():
    print(f"  📦 {container['name']}")

# Créer un conteneur
blob_service.create_container("backups")

# Uploader un fichier
blob_client = blob_service.get_blob_client(container="backups", blob="2026-03-20.tar.gz")
with open("backup.tar.gz", "rb") as data:
    blob_client.upload_blob(data, overwrite=True)

# Télécharger un fichier
with open("/tmp/backup.tar.gz", "wb") as f:
    data = blob_client.download_blob()
    data.readinto(f)

# Lister les blobs
container_client = blob_service.get_container_client("backups")
for blob in container_client.list_blobs():
    print(f"  {blob.name} — {blob.size / 1024 / 1024:.2f} MB")

Compute — Machines virtuelles

from azure.mgmt.compute import ComputeManagementClient
from azure.mgmt.network import NetworkManagementClient
from azure.identity import DefaultAzureCredential

subscription_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
credential = DefaultAzureCredential()
compute_client = ComputeManagementClient(credential, subscription_id)

# Lister les VMs
for vm in compute_client.virtual_machines.list("my-resource-group"):
    print(f"  {vm.name} | {vm.hardware_profile.vm_size} | {vm.location}")

# Démarrer / arrêter une VM
compute_client.virtual_machines.begin_start("my-resource-group", "my-vm").result()
compute_client.virtual_machines.begin_deallocate("my-resource-group", "my-vm").result()

Docker SDK pour Python

Le Docker SDK permet de gérer les conteneurs, images et réseaux Docker depuis Python.

pip install docker

Opérations de base

import docker

client = docker.from_env()

# Informations Docker
info = client.info()
print(f"Docker version : {client.version()['Version']}")
print(f"Conteneurs : {info['Containers']}")
print(f"Images : {info['Images']}")

Gestion des conteneurs

# Lister les conteneurs
for container in client.containers.list(all=True):
    print(f"  {container.short_id} | {container.name:20} | {container.status:10} | {container.image.tags}")

# Lancer un conteneur
container = client.containers.run(
    "nginx:latest",
    name="web-test",
    ports={"80/tcp": 8080},
    detach=True,
    environment={"NGINX_HOST": "example.com"},
    labels={"app": "web", "env": "test"},
    restart_policy={"Name": "unless-stopped"}
)
print(f"✅ Conteneur lancé : {container.id[:12]}")

# Logs d'un conteneur
logs = container.logs(tail=50).decode()
print(logs)

# Logs en streaming
for line in container.logs(stream=True, follow=True):
    print(line.decode().strip())

# Exécuter une commande dans un conteneur
exit_code, output = container.exec_run("nginx -t")
print(f"Test config nginx : {'OK' if exit_code == 0 else 'FAILED'}")

# Arrêter et supprimer
container.stop(timeout=10)
container.remove()

# Stats d'un conteneur
stats = container.stats(stream=False)
cpu_percent = stats["cpu_stats"]["cpu_usage"]["total_usage"]
mem_usage = stats["memory_stats"]["usage"] / 1024 / 1024
print(f"CPU: {cpu_percent}, Memory: {mem_usage:.1f} MB")

Gestion des images

# Lister les images
for image in client.images.list():
    tags = image.tags if image.tags else ["<none>"]
    size_mb = image.attrs["Size"] / 1024 / 1024
    print(f"  {tags[0]:40} {size_mb:8.1f} MB")

# Construire une image
image, logs = client.images.build(
    path=".",
    tag="myapp:latest",
    rm=True,
    buildargs={"APP_VERSION": "2.1.0"}
)

for log in logs:
    if "stream" in log:
        print(log["stream"], end="")

# Pousser une image
client.images.push("registry.example.com/myapp", tag="latest")

# Puller une image
client.images.pull("postgres", tag="16")

# Nettoyer les images non utilisées
pruned = client.images.prune()
print(f"Espace récupéré : {pruned['SpaceReclaimed'] / 1024 / 1024:.1f} MB")

Docker Compose programmatique

import docker
import yaml

def deploy_compose(compose_file):
    """Déploie un stack Docker Compose programmatiquement."""
    client = docker.from_env()

    with open(compose_file) as f:
        compose = yaml.safe_load(f)

    for service_name, config in compose.get("services", {}).items():
        image = config.get("image", f"{service_name}:latest")
        ports = {}
        for port_mapping in config.get("ports", []):
            host, container = str(port_mapping).split(":")
            ports[f"{container}/tcp"] = int(host)

        environment = config.get("environment", {})
        if isinstance(environment, list):
            environment = dict(e.split("=", 1) for e in environment)

        try:
            existing = client.containers.get(service_name)
            existing.stop()
            existing.remove()
        except docker.errors.NotFound:
            pass

        container = client.containers.run(
            image,
            name=service_name,
            ports=ports,
            environment=environment,
            detach=True,
            restart_policy={"Name": "unless-stopped"}
        )
        print(f"  ✅ {service_name} → {container.short_id}")

À toi de jouer

Exercice 1 — Liste tes ressources avec Boto3

Écris un script qui utilise Boto3 pour lister toutes les instances EC2 d’une région avec leur état, IP et tags :

import boto3
ec2 = boto3.client('ec2', region_name='eu-west-1')
response = ec2.describe_instances()
for r in response['Reservations']:
    for i in r['Instances']:
        print(f"{i['InstanceId']} - {i['State']['Name']}")

Exercice 2 — Gère des objets S3

Crée un script qui crée un bucket S3, upload un fichier, liste le contenu du bucket, et génère une URL pré-signée pour le téléchargement. Gère les exceptions proprement.

Exercice 3 — Script multi-cloud

Crée un script cloud_inventory.py qui interroge à la fois AWS (Boto3) et Docker (Docker SDK) pour lister les instances EC2 et les conteneurs Docker en cours d’exécution. Affiche un inventaire unifié au format JSON.

➡️ La suite

Dans le prochain chapitre, on intègre Python avec Terraform, Ansible et Kubernetes pour un workflow d’infrastructure complet.