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--helpgratuits.loggingplutôt queprint: 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=Truetransforme 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èque | Usage |
|---|---|
requests / httpx | Appels HTTP |
PyYAML | Fichiers YAML |
google-cloud-* | API GCP (storage, pubsub, secret-manager) |
boto3 | API AWS |
paramiko / fabric | SSH |
kafka-python / confluent-kafka | Kafka |
pika | RabbitMQ |
psutil | CPU, mémoire, disque, processus |
pytest | Tests |
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_loadet pasyaml.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.