revision.devops

Python pour l'exploitation

On ne te demandera pas de faire du développement applicatif. Le Python d'un admin sert à automatiser : appeler une API, parser des logs ou du JSON, lancer des commandes, produire un rapport.

Structure d'un script propre

#!/usr/bin/env python3
import argparse
import logging
import sys

log = logging.getLogger(__name__)


def main() -> int:
    parser = argparse.ArgumentParser(description="Vérifie l'espace disque")
    parser.add_argument("--seuil", type=int, default=90)
    parser.add_argument("-v", "--verbose", action="store_true")
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )

    log.info("Seuil : %s %%", args.seuil)
    return 0


if __name__ == "__main__":
    sys.exit(main())
  • argparse : arguments et --help gratuits.
  • logging plutôt que print : niveaux, horodatage, redirection facile.
  • sys.exit(code) : code de retour exploitable par cron, la CI, la supervision.
  • if __name__ == "__main__" : le fichier peut aussi être importé sans s'exécuter.

Lancer des commandes

import subprocess

result = subprocess.run(
    ["systemctl", "is-active", "nginx"],
    capture_output=True,
    text=True,
    check=False,
    timeout=10,
)
print(result.returncode, result.stdout.strip())

subprocess.run(["rsync", "-a", src, dst], check=True)   # lève CalledProcessError si échec
  • Passer une liste, pas une chaîne avec shell=True : pas d'injection possible si une variable contient ; rm -rf /.
  • check=True transforme un échec en exception.
  • timeout évite un script bloqué pour toujours.

Fichiers et chemins

from pathlib import Path

logs = Path("/var/log/app")
for f in logs.glob("*.log"):
    if f.stat().st_size > 100 * 1024 * 1024:
        print(f.name, "dépasse 100 Mo")

contenu = Path("config.txt").read_text(encoding="utf-8")
Path("rapport.txt").write_text("OK\n", encoding="utf-8")

with open("gros.log", encoding="utf-8") as fh:
    for ligne in fh:          # ligne par ligne, sans tout charger en mémoire
        if "ERROR" in ligne:
            print(ligne.rstrip())

JSON et YAML

import json

data = json.loads('{"status": "ok", "items": [1, 2]}')
print(data["items"])
print(json.dumps(data, indent=2))

import yaml                  # paquet PyYAML
with open("inventaire.yml") as fh:
    inv = yaml.safe_load(fh) # safe_load, jamais load : load peut exécuter du code

Appeler une API

import os
import requests

token = os.environ["GITLAB_TOKEN"]          # secret depuis l'environnement, jamais en dur
resp = requests.get(
    "https://gitlab.example.com/api/v4/projects",
    headers={"PRIVATE-TOKEN": token},
    params={"per_page": 100},
    timeout=10,
)
resp.raise_for_status()                     # exception si 4xx/5xx
for projet in resp.json():
    print(projet["path_with_namespace"])

Toujours un timeout. Sans lui, requests peut attendre indéfiniment.

Structures utiles

from collections import Counter, defaultdict

ips = Counter()
with open("access.log") as fh:
    for ligne in fh:
        ips[ligne.split()[0]] += 1
for ip, n in ips.most_common(10):
    print(f"{n:>6} {ip}")

par_code = defaultdict(list)
par_code[500].append("/api/orders")

Compréhensions :

actifs = [s for s in services if est_actif(s)]
tailles = {f.name: f.stat().st_size for f in Path(".").iterdir() if f.is_file()}

Expressions régulières

import re

motif = re.compile(r'^(\S+) .* "(\w+) (\S+) [^"]*" (\d{3})')
m = motif.match(ligne)
if m:
    ip, methode, url, code = m.groups()

Gestion d'erreurs

try:
    resp = requests.get(url, timeout=5)
    resp.raise_for_status()
except requests.Timeout:
    log.error("Timeout sur %s", url)
    return 2
except requests.HTTPError as e:
    log.error("HTTP %s sur %s", e.response.status_code, url)
    return 1

Attraper les exceptions précises. Un except Exception: pass cache tous les problèmes.

Dates

from datetime import datetime, timedelta, timezone

maintenant = datetime.now(timezone.utc)
il_y_a_7j = maintenant - timedelta(days=7)
print(maintenant.isoformat())

Environnements et dépendances

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

Un venv par projet, pour ne pas casser le Python du système (dont dépendent dnf ou apt). Outils plus récents : uv, très rapide, et pipx pour installer des outils CLI isolés.

Bibliothèques utiles en exploitation

BibliothèqueUsage
requests / httpxAppels HTTP
PyYAMLFichiers YAML
google-cloud-*API GCP (storage, pubsub, secret-manager)
boto3API AWS
paramiko / fabricSSH
kafka-python / confluent-kafkaKafka
pikaRabbitMQ
psutilCPU, mémoire, disque, processus
pytestTests

Exemple avec le SDK GCP :

from google.cloud import secretmanager

client = secretmanager.SecretManagerServiceClient()
nom = "projects/mon-projet/secrets/db-password/versions/latest"
valeur = client.access_secret_version(name=nom).payload.data.decode()

Les identifiants viennent des Application Default Credentials (compte de service de la VM, gcloud auth application-default login, ou Workload Identity). Rien à mettre dans le code.

Questions d'entretien à préparer

  • Pourquoi éviter shell=True ?
  • Comment lire un fichier de 10 Go sans saturer la mémoire ? (itérer ligne par ligne)
  • Pourquoi yaml.safe_load et pas yaml.load ?
  • Où mets-tu le token d'API utilisé par ton script ?
  • Quand passes-tu de Bash à Python ?
  • Écris un script qui compte les codes HTTP d'un access log.