Python 3.3 : yield from, venv et les namespace packages

# Pipeline de traitement de fichiers journaux

def lire_lignes(chemin):
    """Générateur qui lit un fichier ligne par ligne."""
    with open(chemin) as f:
        for ligne in f:
            yield ligne.strip()

def filtrer_erreurs(lignes):
    """Ne conserve que les lignes contenant ERROR."""
    for ligne in lignes:
        if "ERROR" in ligne:
            yield ligne

def extraire_messages(lignes):
    """Extrait le message après le niveau de log."""
    for ligne in lignes:
        parties = ligne.split(" - ", maxsplit=2)
        if len(parties) >= 3:
            yield parties[2]

def pipeline_erreurs(chemins):
    """Pipeline complet : lit plusieurs fichiers et extrait les erreurs.

    Grâce à yield from, chaque étape délègue naturellement
    au sous-générateur sans boucle explicite.
    """
    for chemin in chemins:
        lignes = lire_lignes(chemin)
        erreurs = filtrer_erreurs(lignes)
        yield from extraire_messages(erreurs)

# Utilisation
fichiers = ["app.log", "worker.log", "api.log"]
for message in pipeline_erreurs(fichiers):
    print(f"[ERREUR] {message}")

# Autre exemple : aplatir une structure imbriquée
def aplatir(structure):
    """Aplatit récursivement une structure imbriquée."""
    for element in structure:
        if isinstance(element, (list, tuple)):
            yield from aplatir(element)
        else:
            yield element

donnees = [1, [2, 3], [4, [5, 6]], 7]
print(list(aplatir(donnees)))  # [1, 2, 3, 4, 5, 6, 7]

# Création d'un environnement virtuel depuis le terminal
# python -m venv mon_projet_env

# Activation (Linux/macOS)
# source mon_projet_env/bin/activate

# Activation (Windows)
# mon_projet_env\Scripts\activate

# Vérification que l'environnement est actif
import sys
import os

def verifier_environnement():
    """Affiche les informations sur l'environnement Python actif."""
    venv_path = os.environ.get("VIRTUAL_ENV")
    if venv_path:
        print(f"Environnement virtuel actif : {venv_path}")
        print(f"Exécutable Python : {sys.executable}")
        print(f"Chemin des paquets :")
        for chemin in sys.path:
            if "site-packages" in chemin:
                print(f"  {chemin}")
    else:
        print("Aucun environnement virtuel actif.")

verifier_environnement()

# Création programmatique d'un environnement
import venv

constructeur = venv.EnvBuilder(with_pip=True)
constructeur.create("test_env")
print("Environnement 'test_env' créé avec pip intégré.")

# Structure sur disque :
#
# repertoire_a/
#   monframework/
#     core/
#       __init__.py
#       moteur.py
#
# repertoire_b/
#   monframework/
#     plugins/
#       __init__.py
#       export_csv.py
#
# Pas de __init__.py dans monframework/ !
# Python 3.3 fusionne automatiquement les deux parties.

# Dans le code, on importe normalement :
# import monframework.core.moteur
# import monframework.plugins.export_csv

# Vérification qu'un paquet est un namespace package
import importlib

def inspecter_paquet(nom):
    """Affiche si un paquet est un namespace package."""
    try:
        module = importlib.import_module(nom)
        if hasattr(module, "__path__"):
            chemins = list(module.__path__)
            if hasattr(module, "__file__") and module.__file__:
                print(f"{nom} : paquet classique ({module.__file__})")
            else:
                print(f"{nom} : namespace package")
                for chemin in chemins:
                    print(f"  -> {chemin}")
        else:
            print(f"{nom} : module simple")
    except ImportError:
        print(f"{nom} : introuvable")

inspecter_paquet("json")  # paquet classique
inspecter_paquet("os")    # module simple

import faulthandler
import sys
import threading

# Activer faulthandler pour capturer les segfaults
faulthandler.enable()

# Afficher la pile d'appels de tous les threads
# (utile pour débugger un programme qui semble bloqué)
def tache_longue():
    """Simule une tâche bloquante."""
    import time
    time.sleep(30)

fil = threading.Thread(target=tache_longue, name="TravailFond")
fil.daemon = True
fil.start()

# Afficher l'état de tous les threads
faulthandler.dump_traceback(file=sys.stderr, all_threads=True)

# Déclencher un dump après un délai (watchdog)
# Si le programme ne répond pas dans les 10 secondes,
# la pile d'appels sera affichée automatiquement.
faulthandler.dump_traceback_later(
    timeout=10,
    repeat=False,
    file=sys.stderr,
)

# On peut aussi l'activer via la ligne de commande :
# python -X faulthandler mon_script.py
#
# Ou via une variable d'environnement :
# PYTHONFAULTHANDLER=1 python mon_script.py

# Annuler le watchdog si tout se passe bien
faulthandler.cancel_dump_traceback_later()