# Fichier: python_cheats/cheatsheets/subprocess_complete.txt
# Python Subprocess - Module d'Exécution de Processus Complet



[OK] IMPORT ET CONCEPTS DE BASE


[OK] IMPORT
    import subprocess
    from subprocess import (
        run, call, check_call, check_output,
        Popen, PIPE, STDOUT, DEVNULL,
        TimeoutExpired, CalledProcessError,
        CompletedProcess
    )
    import shlex  # Pour parser les commandes
    import os
    import sys

[OK] CONCEPTS CLÉS
    # subprocess remplace os.system(), os.spawn*(), os.popen*(), commands.*
    # Utilise toujours subprocess pour exécuter des commandes externes
    
    # Principales fonctions:
    # - subprocess.run()          -> Fonction recommandée (Python 3.5+)
    # - subprocess.Popen()        -> Contrôle bas niveau, avancé
    # - subprocess.call()         -> Ancienne méthode (avant 3.5)
    # - subprocess.check_call()   -> Ancienne méthode avec vérification
    # - subprocess.check_output() -> Ancienne méthode pour capturer sortie


[OK] SUBPROCESS.RUN() - MÉTHODE RECOMMANDÉE


[OK] RUN BASIQUE
    # Commande simple
    result = subprocess.run(['ls', '-l'])
    
    # Avec arguments séparés
    result = subprocess.run(['echo', 'Hello', 'World'])
    
    # Depuis string (nécessite shell=True, [ATTENTION] attention sécurité!)
    result = subprocess.run('ls -l', shell=True)
    
    # Meilleure pratique: utiliser shlex.split()
    import shlex
    cmd = 'ls -l /tmp'
    result = subprocess.run(shlex.split(cmd))

[OK] OBJET COMPLETEDPROCESS
    result = subprocess.run(['ls', '-l'])
    
    # Attributs disponibles:
    print(result.args)         # Commande exécutée: ['ls', '-l']
    print(result.returncode)   # Code de retour: 0 (succès), != 0 (erreur)
    print(result.stdout)       # Sortie standard (si capture=True)
    print(result.stderr)       # Sortie d'erreur (si capture=True)
    
    # Vérifier le succès
    if result.returncode == 0:
        print("Commande réussie")
    else:
        print("Commande échouée")

[OK] CAPTURER LA SORTIE
    # Capturer stdout
    result = subprocess.run(['ls', '-l'], 
                           capture_output=True,  # Python 3.7+
                           text=True)            # Decode en string
    print(result.stdout)
    
    # Alternative (Python 3.5-3.6)
    result = subprocess.run(['ls', '-l'],
                           stdout=subprocess.PIPE,
                           stderr=subprocess.PIPE,
                           universal_newlines=True)  # Équivalent de text=True
    
    # Capturer en bytes
    result = subprocess.run(['ls', '-l'],
                           capture_output=True)
    print(result.stdout.decode('utf-8'))

[OK] GESTION DES ERREURS
    # Lever exception si erreur (check=True)
    try:
        result = subprocess.run(['ls', '/nonexistent'],
                               check=True,
                               capture_output=True,
                               text=True)
    except subprocess.CalledProcessError as e:
        print(f"Commande échouée avec code: {e.returncode}")
        print(f"Sortie d'erreur: {e.stderr}")
        print(f"Sortie standard: {e.stdout}")
    
    # Sans check=True (gestion manuelle)
    result = subprocess.run(['ls', '/nonexistent'],
                           capture_output=True,
                           text=True)
    if result.returncode != 0:
        print(f"Erreur: {result.stderr}")

[OK] TIMEOUT
    # Timeout en secondes
    try:
        result = subprocess.run(['sleep', '10'],
                               timeout=5,
                               capture_output=True)
    except subprocess.TimeoutExpired as e:
        print(f"Commande timeout après {e.timeout} secondes")
        print(f"Sortie partielle: {e.output}")
        print(f"Stderr partiel: {e.stderr}")

[OK] RÉPERTOIRE DE TRAVAIL ET ENVIRONNEMENT
    # Changer le répertoire de travail
    result = subprocess.run(['ls'], cwd='/tmp')
    
    # Variables d'environnement personnalisées
    env = os.environ.copy()
    env['MY_VAR'] = 'my_value'
    env['PATH'] = '/custom/path:' + env['PATH']
    result = subprocess.run(['printenv', 'MY_VAR'], 
                           env=env,
                           capture_output=True,
                           text=True)
    print(result.stdout)  # my_value

[OK] INPUT (STDIN)
    # Passer input via string
    result = subprocess.run(['cat'],
                           input='Hello\nWorld\n',
                           capture_output=True,
                           text=True)
    print(result.stdout)  # Hello\nWorld\n
    
    # Passer input via bytes
    result = subprocess.run(['cat'],
                           input=b'Binary data',
                           capture_output=True)
    
    # Lire depuis fichier
    with open('input.txt', 'r') as f:
        result = subprocess.run(['grep', 'pattern'],
                               stdin=f,
                               capture_output=True,
                               text=True)


[OK] POPEN - CONTRÔLE AVANCÉ


[OK] POPEN BASIQUE
    # Créer un processus
    process = subprocess.Popen(['ls', '-l'])
    
    # Attendre la fin
    process.wait()
    
    # Obtenir le code de retour
    returncode = process.returncode
    print(f"Code de retour: {returncode}")

[OK] CAPTURER STDOUT ET STDERR
    # Capturer stdout
    process = subprocess.Popen(['ls', '-l'],
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              text=True)
    
    # Lire la sortie
    stdout, stderr = process.communicate()
    print(f"Sortie: {stdout}")
    print(f"Erreur: {stderr}")
    print(f"Code retour: {process.returncode}")
    
    # Lire ligne par ligne (attention aux deadlocks!)
    process = subprocess.Popen(['ls', '-l'],
                              stdout=subprocess.PIPE,
                              text=True)
    for line in process.stdout:
        print(line.strip())
    process.wait()

[OK] POPEN AVEC INPUT
    # Envoyer input au processus
    process = subprocess.Popen(['cat'],
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              text=True)
    
    stdout, stderr = process.communicate(input='Hello\nWorld\n')
    print(stdout)

[OK] INTERACTION EN TEMPS RÉEL
    # Écrire et lire progressivement
    process = subprocess.Popen(['python', '-u'],  # -u pour unbuffered
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              text=True,
                              bufsize=1)  # Line buffered
    
    # Envoyer commande
    process.stdin.write('print("Hello")\n')
    process.stdin.flush()
    
    # Lire réponse
    output = process.stdout.readline()
    print(output.strip())
    
    # Fermer proprement
    process.stdin.close()
    process.wait()

[OK] PROCESSUS EN ARRIÈRE-PLAN
    # Lancer et continuer sans attendre
    process = subprocess.Popen(['long_running_command'])
    print(f"Processus lancé avec PID: {process.pid}")
    
    # Vérifier si toujours en cours
    if process.poll() is None:
        print("Processus toujours en cours")
    else:
        print(f"Processus terminé avec code: {process.returncode}")
    
    # Attendre avec timeout
    try:
        returncode = process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        print("Processus toujours en cours après 5 secondes")
        process.kill()  # Forcer l'arrêt

[OK] TERMINER UN PROCESSUS
    process = subprocess.Popen(['sleep', '100'])
    
    # Envoi signal SIGTERM (arrêt propre)
    process.terminate()
    
    # Envoi signal SIGKILL (arrêt forcé)
    process.kill()
    
    # Attendre la fin après terminate/kill
    process.wait()
    
    # Envoyer signal personnalisé (Unix seulement)
    import signal
    process.send_signal(signal.SIGUSR1)


[OK] PIPES ET REDIRECTIONS


[OK] PIPE ENTRE COMMANDES
    # Équivalent de: ls -l | grep py | wc -l
    
    # Méthode 1: Manuelle avec Popen
    p1 = subprocess.Popen(['ls', '-l'],
                         stdout=subprocess.PIPE)
    p2 = subprocess.Popen(['grep', 'py'],
                         stdin=p1.stdout,
                         stdout=subprocess.PIPE)
    p1.stdout.close()  # Permettre p1 de recevoir SIGPIPE si p2 termine
    p3 = subprocess.Popen(['wc', '-l'],
                         stdin=p2.stdout,
                         stdout=subprocess.PIPE,
                         text=True)
    p2.stdout.close()
    
    output, _ = p3.communicate()
    print(output)
    
    # Méthode 2: Avec run() (Python 3.7+)
    p1 = subprocess.run(['ls', '-l'],
                       capture_output=True)
    p2 = subprocess.run(['grep', 'py'],
                       input=p1.stdout,
                       capture_output=True)
    p3 = subprocess.run(['wc', '-l'],
                       input=p2.stdout,
                       capture_output=True,
                       text=True)
    print(p3.stdout)

[OK] REDIRIGER VERS FICHIER
    # Rediriger stdout vers fichier
    with open('output.txt', 'w') as f:
        subprocess.run(['ls', '-l'], stdout=f)
    
    # Rediriger stderr vers fichier
    with open('errors.txt', 'w') as f:
        subprocess.run(['ls', '/nonexistent'], stderr=f)
    
    # Rediriger stdout et stderr vers le même fichier
    with open('combined.txt', 'w') as f:
        subprocess.run(['command'],
                      stdout=f,
                      stderr=subprocess.STDOUT)
    
    # Append mode
    with open('output.txt', 'a') as f:
        subprocess.run(['echo', 'new line'], stdout=f)

[OK] LIRE DEPUIS FICHIER
    # Lire stdin depuis fichier
    with open('input.txt', 'r') as f:
        result = subprocess.run(['sort'],
                               stdin=f,
                               capture_output=True,
                               text=True)
    print(result.stdout)

[OK] DEVNULL (SUPPRIMER SORTIE)
    # Ignorer stdout
    subprocess.run(['command'], stdout=subprocess.DEVNULL)
    
    # Ignorer stderr
    subprocess.run(['command'], stderr=subprocess.DEVNULL)
    
    # Ignorer les deux
    subprocess.run(['command'],
                  stdout=subprocess.DEVNULL,
                  stderr=subprocess.DEVNULL)

[OK] REDIRIGER STDERR VERS STDOUT
    # stderr vers stdout
    result = subprocess.run(['command'],
                           stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT,
                           text=True)
    print(result.stdout)  # Contient stdout + stderr


[OK] SHELL=TRUE - UTILISATION ET DANGERS


[OK] QUAND UTILISER SHELL=TRUE
    # Commandes avec pipes, redirections, wildcards
    subprocess.run('ls *.txt | wc -l', shell=True)
    
    # Variables d'environnement shell
    subprocess.run('echo $HOME', shell=True)
    
    # Commandes avec redirections
    subprocess.run('command > output.txt 2>&1', shell=True)
    
    # Commandes shell builtin
    subprocess.run('cd /tmp && pwd', shell=True)

[OK] DANGERS DE SHELL=TRUE (INJECTION DE COMMANDES)
    # [ATTENTION] TRÈS DANGEREUX - NE JAMAIS FAIRE ÇA!
    user_input = input("Enter filename: ")  # evil.txt; rm -rf /
    subprocess.run(f'cat {user_input}', shell=True)  # INJECTION!
    
    # [OK] SOLUTION: Ne pas utiliser shell=True
    subprocess.run(['cat', user_input])
    
    # Si shell=True nécessaire, utiliser shlex.quote()
    import shlex
    safe_input = shlex.quote(user_input)
    subprocess.run(f'cat {safe_input}', shell=True)

[OK] ALTERNATIVES À SHELL=TRUE
    # Au lieu de: subprocess.run('ls *.txt', shell=True)
    # Utiliser glob:
    import glob
    files = glob.glob('*.txt')
    subprocess.run(['ls'] + files)
    
    # Au lieu de: subprocess.run('echo $HOME', shell=True)
    # Utiliser os.environ:
    import os
    subprocess.run(['echo', os.environ['HOME']])
    
    # Au lieu de pipes shell, utiliser pipes Python (voir section PIPES)


[OK] ENCODAGE ET CARACTÈRES SPÉCIAUX


[OK] GESTION DE L'ENCODAGE
    # Sortie en string (text=True)
    result = subprocess.run(['ls'],
                           capture_output=True,
                           text=True)
    print(type(result.stdout))  # <class 'str'>
    
    # Sortie en bytes (défaut)
    result = subprocess.run(['ls'],
                           capture_output=True)
    print(type(result.stdout))  # <class 'bytes'>
    
    # Spécifier encodage
    result = subprocess.run(['ls'],
                           capture_output=True,
                           text=True,
                           encoding='utf-8')
    
    # Gestion erreurs d'encodage
    result = subprocess.run(['command'],
                           capture_output=True,
                           text=True,
                           encoding='utf-8',
                           errors='replace')  # ou 'ignore', 'backslashreplace'

[OK] CARACTÈRES SPÉCIAUX ET ÉCHAPPEMENT
    # Caractères spéciaux dans arguments
    # [OK] Avec liste (pas d'échappement nécessaire)
    subprocess.run(['echo', 'hello $USER & world'])
    
    # [ATTENTION] Avec shell=True (nécessite échappement)
    import shlex
    arg = 'hello $USER & world'
    subprocess.run(f'echo {shlex.quote(arg)}', shell=True)

[OK] NEWLINES
    # Gestion des newlines
    result = subprocess.run(['printf', 'line1\nline2\r\nline3'],
                           capture_output=True,
                           text=True,
                           universal_newlines=True)  # Normalise \n, \r\n, \r -> \n
    
    print(repr(result.stdout))


[OK] SUBPROCESS AVEC DIFFÉRENTS OS


[OK] COMMANDES CROSS-PLATFORM
    import platform
    
    # Détecter l'OS
    if platform.system() == 'Windows':
        subprocess.run(['dir'], shell=True)
    else:  # Linux, macOS
        subprocess.run(['ls', '-l'])
    
    # Commande Python cross-platform
    subprocess.run([sys.executable, 'script.py'])

[OK] WINDOWS SPÉCIFIQUE
    # Cacher fenêtre de console (Windows)
    if platform.system() == 'Windows':
        startupinfo = subprocess.STARTUPINFO()
        startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
        
        subprocess.run(['command'],
                      startupinfo=startupinfo)
    
    # Créer nouveau groupe de processus (Windows)
    subprocess.Popen(['command'],
                    creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)

[OK] UNIX/LINUX SPÉCIFIQUE
    # Préserver les descripteurs de fichiers (Unix)
    subprocess.Popen(['command'],
                    pass_fds=(3, 4, 5))
    
    # Changer utilisateur (nécessite root)
    import pwd
    def demote(user_uid, user_gid):
        def set_ids():
            os.setgid(user_gid)
            os.setuid(user_uid)
        return set_ids
    
    pw_record = pwd.getpwnam('username')
    subprocess.Popen(['command'],
                    preexec_fn=demote(pw_record.pw_uid, pw_record.pw_gid))
    
    # Changer répertoire avant exec (Unix)
    subprocess.Popen(['command'],
                    preexec_fn=lambda: os.chdir('/new/dir'))


[OK] ANCIENNES FONCTIONS (PRÉ-3.5)


[OK] SUBPROCESS.CALL()
    # Retourne le code de retour
    # Équivalent à: run(...).returncode
    returncode = subprocess.call(['ls', '-l'])
    print(returncode)  # 0 si succès

[OK] SUBPROCESS.CHECK_CALL()
    # Comme call() mais lève CalledProcessError si échec
    # Équivalent à: run(..., check=True)
    try:
        subprocess.check_call(['ls', '-l'])
    except subprocess.CalledProcessError as e:
        print(f"Erreur: {e.returncode}")

[OK] SUBPROCESS.CHECK_OUTPUT()
    # Retourne stdout, lève exception si échec
    # Équivalent à: run(..., check=True, capture_output=True).stdout
    try:
        output = subprocess.check_output(['ls', '-l'],
                                        text=True)
        print(output)
    except subprocess.CalledProcessError as e:
        print(f"Erreur: {e.output}")

[OK] SUBPROCESS.GETOUTPUT() ET GETSTATUSOUTPUT()
    # Unix seulement, remplace commands.getoutput()
    output = subprocess.getoutput('ls -l')
    
    status, output = subprocess.getstatusoutput('ls -l')
    print(f"Status: {status}")
    print(f"Output: {output}")


[OK] PATTERNS AVANCÉS


[OK] EXÉCUTER PLUSIEURS COMMANDES EN PARALLÈLE
    import concurrent.futures
    
    commands = [
        ['command1'],
        ['command2'],
        ['command3']
    ]
    
    def run_command(cmd):
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True)
        return result.stdout
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        results = list(executor.map(run_command, commands))
    
    for i, output in enumerate(results):
        print(f"Commande {i+1}: {output}")

[OK] WRAPPER AVEC RETRY
    import time
    
    def run_with_retry(cmd, max_attempts=3, delay=1):
        """Exécute commande avec retry en cas d'échec"""
        for attempt in range(max_attempts):
            try:
                result = subprocess.run(cmd,
                                       capture_output=True,
                                       text=True,
                                       check=True,
                                       timeout=10)
                return result
            
            except (subprocess.CalledProcessError, 
                    subprocess.TimeoutExpired) as e:
                if attempt < max_attempts - 1:
                    print(f"Tentative {attempt + 1} échouée. Retry...")
                    time.sleep(delay)
                    delay *= 2  # Exponential backoff
                else:
                    raise
    
    result = run_with_retry(['curl', 'https://api.example.com'])

[OK] LOGGER LES COMMANDES
    import logging
    
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)
    
    def run_logged(cmd, **kwargs):
        """Exécute commande avec logging"""
        logger.info(f"Exécution: {' '.join(cmd)}")
        
        start_time = time.time()
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True,
                               **kwargs)
        elapsed = time.time() - start_time
        
        logger.info(f"Durée: {elapsed:.2f}s, Code: {result.returncode}")
        
        if result.returncode != 0:
            logger.error(f"Erreur: {result.stderr}")
        
        return result
    
    result = run_logged(['ls', '-l'], cwd='/tmp')

[OK] WRAPPER CLASSE POUR COMMANDES
    class Command:
        """Wrapper orienté objet pour subprocess"""
        
        def __init__(self, cmd, cwd=None, env=None):
            self.cmd = cmd if isinstance(cmd, list) else shlex.split(cmd)
            self.cwd = cwd
            self.env = env
            self.result = None
        
        def run(self, timeout=None, input=None):
            """Exécute la commande"""
            self.result = subprocess.run(
                self.cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                input=input,
                cwd=self.cwd,
                env=self.env
            )
            return self
        
        @property
        def stdout(self):
            return self.result.stdout if self.result else None
        
        @property
        def stderr(self):
            return self.result.stderr if self.result else None
        
        @property
        def returncode(self):
            return self.result.returncode if self.result else None
        
        @property
        def success(self):
            return self.returncode == 0
        
        def __repr__(self):
            return f"Command({' '.join(self.cmd)})"
    
    # Utilisation
    cmd = Command('ls -l /tmp')
    cmd.run()
    if cmd.success:
        print(cmd.stdout)
    else:
        print(f"Erreur: {cmd.stderr}")

[OK] EXÉCUTER AVEC PROGRESS BAR
    from tqdm import tqdm
    
    def run_with_progress(cmd, desc="Exécution"):
        """Exécute commande avec progress bar pour stdout"""
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1
        )
        
        output_lines = []
        with tqdm(desc=desc, unit=' lines') as pbar:
            for line in process.stdout:
                output_lines.append(line)
                pbar.update(1)
        
        process.wait()
        stderr = process.stderr.read()
        
        return output_lines, stderr, process.returncode
    
    lines, errors, code = run_with_progress(['ls', '-lR', '/'])


[OK] SÉCURITÉ ET BONNES PRATIQUES


[OK] RÈGLES DE SÉCURITÉ
    # [OK] BON: Liste d'arguments
    subprocess.run(['grep', user_pattern, filename])
    
    # [ATTENTION] MAUVAIS: String avec shell=True
    subprocess.run(f'grep {user_pattern} {filename}', shell=True)
    
    # [OK] BON: Si shell nécessaire, utiliser shlex.quote()
    import shlex
    cmd = f'grep {shlex.quote(user_pattern)} {shlex.quote(filename)}'
    subprocess.run(cmd, shell=True)
    
    # [OK] BON: Valider input utilisateur
    import re
    if re.match(r'^[a-zA-Z0-9_-]+$', user_input):
        subprocess.run(['command', user_input])
    else:
        raise ValueError("Input invalide")

[OK] LIMITER LES RESSOURCES (UNIX)
    import resource
    
    def limit_resources():
        """Limite CPU et mémoire pour le processus enfant"""
        # Limite CPU à 10 secondes
        resource.setrlimit(resource.RLIMIT_CPU, (10, 10))
        
        # Limite mémoire à 100MB
        resource.setrlimit(resource.RLIMIT_AS, (100*1024*1024, 100*1024*1024))
    
    try:
        subprocess.run(['command'],
                      preexec_fn=limit_resources,
                      timeout=15)
    except subprocess.TimeoutExpired:
        print("Processus a dépassé les limites")

[OK] ISOLATION ET SANDBOX
    # Exécuter dans un répertoire temporaire isolé
    import tempfile
    
    with tempfile.TemporaryDirectory() as tmpdir:
        subprocess.run(['command'],
                      cwd=tmpdir,
                      env={'PATH': '/usr/bin:/bin'})  # Env minimal

[OK] CHECKLIST DE SÉCURITÉ
    """
    [OK] Utiliser liste d'arguments au lieu de string
    [OK] Éviter shell=True autant que possible
    [OK] Valider et nettoyer les inputs utilisateur
    [OK] Utiliser shlex.quote() si shell=True nécessaire
    [OK] Utiliser timeout pour éviter blocages
    [OK] Limiter les variables d'environnement transmises
    [OK] Vérifier les codes de retour
    [OK] Capturer et logger stderr
    [OK] Utiliser check=True pour détecter erreurs
    [OK] Ne jamais exécuter avec privilèges élevés si évitable
    """


[OK] DÉBOGAGE ET TROUBLESHOOTING


[OK] AFFICHER LA COMMANDE EXACTE
    import shlex
    
    cmd = ['ls', '-l', '/tmp']
    print(f"Commande: {shlex.join(cmd)}")  # Python 3.8+
    
    # Python < 3.8
    print(f"Commande: {' '.join(shlex.quote(arg) for arg in cmd)}")

[OK] MODE VERBOSE
    def run_verbose(cmd, **kwargs):
        """Exécute avec output détaillé"""
        print(f">>> Exécution: {' '.join(cmd)}")
        print(f">>> CWD: {kwargs.get('cwd', os.getcwd())}")
        print(f">>> ENV modifié: {kwargs.get('env', {}).keys()}")
        
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True,
                               **kwargs)
        
        print(f">>> Code retour: {result.returncode}")
        print(f">>> Stdout ({len(result.stdout)} chars):")
        print(result.stdout)
        print(f">>> Stderr ({len(result.stderr)} chars):")
        print(result.stderr)
        
        return result

[OK] GESTION DES ZOMBIES
    import signal
    
    # Éviter les processus zombies
    signal.signal(signal.SIGCHLD, signal.SIG_IGN)
    
    # Ou attendre explicitement
    process = subprocess.Popen(['command'])
    # ... faire autre chose ...
    process.wait()  # Nettoyer le processus

[OK] DÉTECTER LES DEADLOCKS
    """
    Deadlocks courants:
    
    1. Lire stdout sans PIPE:
       process = Popen(['command'])
       process.stdout.read()  # [X] Deadlock! stdout n'est pas PIPE
    
    2. Buffer plein:
       process = Popen(['command'], stdout=PIPE)
       # Commande génère beaucoup de données
       process.wait()  # [X] Deadlock! Buffer stdout plein
       
       Solution: Utiliser communicate()
       stdout, stderr = process.communicate()
    
    3. Lire stdout et stderr séparément:
       out = process.stdout.read()  # Peut bloquer
       err = process.stderr.read()  # Si stderr remplit buffer
       
       Solution: Utiliser communicate() ou threads
    """


[OK] EXEMPLES PRATIQUES


[OK] EXÉCUTER SCRIPT PYTHON
    # Exécuter script Python
    subprocess.run([sys.executable, 'script.py'])
    
    # Avec arguments
    subprocess.run([sys.executable, 'script.py', '--arg1', 'value'])
    
    # Capturer sortie
    result = subprocess.run([sys.executable, 'script.py'],
                           capture_output=True,
                           text=True)

[OK] EXÉCUTER COMMANDES GIT
    def git_command(args, cwd='.'):
        """Wrapper pour commandes git"""
        cmd = ['git'] + args
        result = subprocess.run(cmd,
                               cwd=cwd,
                               capture_output=True,
                               text=True,
                               check=True)
        return result.stdout.strip()
    
    # Exemples
    status = git_command(['status', '--short'])
    branch = git_command(['branch', '--show-current'])
    commit = git_command(['log', '-1', '--pretty=%H'])

[OK] COMPRESSION/DÉCOMPRESSION
    # Compresser fichier
    with open('input.txt', 'rb') as f_in:
        with open('output.gz', 'wb') as f_out:
            subprocess.run(['gzip', '-c'],
                          stdin=f_in,
                          stdout=f_out)
    
    # Décompresser
    subprocess.run(['gunzip', 'output.gz'])

[OK] CONVERTIR FICHIERS
    # PDF vers texte (nécessite pdftotext)
    subprocess.run(['pdftotext', 'document.pdf', 'output.txt'])
    
    # Image resize (nécessite ImageMagick)
    subprocess.run(['convert', 'input.jpg',
                   '-resize', '800x600',
                   'output.jpg'])
    
    # Vidéo conversion (nécessite ffmpeg)
    subprocess.run(['ffmpeg',
                   '-i', 'input.mp4',
                   '-c:v', 'libx264',
                   'output.mp4'])

[OK] MONITORING SYSTÈME
    # CPU usage
    result = subprocess.run(['ps', 'aux'],
                           capture_output=True,
                           text=True)
    
    # Disk space
    result = subprocess.run(['df', '-h'],
                           capture_output=True,
                           text=True)
    print(result.stdout)
    
    # Network stats
    result = subprocess.run(['netstat', '-an'],
                           capture_output=True,
                           text=True)

[OK] BACKUP ET SYNC
    # Rsync
    subprocess.run(['rsync',
                   '-avz',
                   '--progress',
                   '/source/path/',
                   '/dest/path/'])
    
    # Tar archive
    subprocess.run(['tar',
                   '-czf',
                   'backup.tar.gz',
                   '/path/to/backup'])
    
    # Extract tar
    subprocess.run(['tar',
                   '-xzf',
                   'backup.tar.gz',
                   '-C',
                   '/extract/path'])

[OK] BASE DE DONNÉES
    # MySQL dump
    subprocess.run(['mysqldump',
                   '-u', 'username',
                   '-p' + password,
                   'database_name'],
                  stdout=open('backup.sql', 'w'))
    
    # PostgreSQL dump
    subprocess.run(['pg_dump',
                   '-U', 'username',
                   'database_name'],
                  stdout=open('backup.sql', 'w'),
                  env={'PGPASSWORD': password})
    
    # Restore
    with open('backup.sql', 'r') as f:
        subprocess.run(['mysql',
                       '-u', 'username',
                       '-p' + password,
                       'database_name'],
                      stdin=f)

[OK] TESTS ET CI/CD
    def run_tests():
        """Exécute suite de tests"""
        tests = [
            ['pytest', 'tests/'],
            ['flake8', '.'],
            ['mypy', 'src/']
        ]
        
        results = {}
        for test in tests:
            result = subprocess.run(test,
                                   capture_output=True,
                                   text=True)
            results[test[0]] = result.returncode == 0
        
        return all(results.values()), results
    
    success, results = run_tests()
    if success:
        print("[OK] Tous les tests passent")
    else:
        print("[X] Tests échoués:", 
              [k for k, v in results.items() if not v])

[OK] DOCKER OPERATIONS
    # Build image
    subprocess.run(['docker', 'build',
                   '-t', 'myapp:latest',
                   '.'])
    
    # Run container
    subprocess.run(['docker', 'run',
                   '-d',
                   '--name', 'mycontainer',
                   '-p', '8080:80',
                   'myapp:latest'])
    
    # Get container logs
    result = subprocess.run(['docker', 'logs', 'mycontainer'],
                           capture_output=True,
                           text=True)
    print(result.stdout)
    
    # Stop and remove
    subprocess.run(['docker', 'stop', 'mycontainer'])
    subprocess.run(['docker', 'rm', 'mycontainer'])

[OK] SSH ET REMOTE COMMANDS
    # SSH command
    subprocess.run(['ssh',
                   'user@remote-host',
                   'ls -l /tmp'])
    
    # SSH avec options
    subprocess.run(['ssh',
                   '-i', '/path/to/key',
                   '-p', '2222',
                   'user@host',
                   'command'])
    
    # SCP file transfer
    subprocess.run(['scp',
                   'local_file.txt',
                   'user@remote:/remote/path/'])
    
    # SCP download
    subprocess.run(['scp',
                   'user@remote:/remote/file.txt',
                   './local_path/'])


[OK] COMMUNICATION INTER-PROCESSUS


[OK] BIDIRECTIONAL COMMUNICATION
    import threading
    
    def read_output(process):
        """Thread pour lire sortie en continu"""
        for line in process.stdout:
            print(f"Output: {line.strip()}")
    
    # Processus interactif
    process = subprocess.Popen(['python', '-u', '-i'],
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              text=True,
                              bufsize=1)
    
    # Thread pour lire output
    output_thread = threading.Thread(target=read_output, args=(process,))
    output_thread.daemon = True
    output_thread.start()
    
    # Envoyer commandes
    process.stdin.write('print("Hello")\n')
    process.stdin.flush()
    
    time.sleep(1)
    
    process.stdin.write('exit()\n')
    process.stdin.flush()
    process.wait()

[OK] QUEUE POUR COMMUNICATION
    from queue import Queue
    import threading
    
    def process_worker(queue):
        """Worker qui exécute commandes depuis queue"""
        while True:
            cmd = queue.get()
            if cmd is None:
                break
            
            result = subprocess.run(cmd,
                                   capture_output=True,
                                   text=True)
            print(f"Commande {cmd[0]} terminée: {result.returncode}")
            queue.task_done()
    
    # Créer queue et workers
    q = Queue()
    workers = []
    for i in range(4):
        t = threading.Thread(target=process_worker, args=(q,))
        t.start()
        workers.append(t)
    
    # Ajouter tâches
    commands = [['ls'], ['pwd'], ['date'], ['whoami']]
    for cmd in commands:
        q.put(cmd)
    
    # Attendre fin
    q.join()
    
    # Arrêter workers
    for _ in workers:
        q.put(None)
    for t in workers:
        t.join()

[OK] SHARED MEMORY (MULTIPROCESSING)
    from multiprocessing import Process, Queue
    
    def run_in_process(cmd, result_queue):
        """Exécute commande dans processus séparé"""
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True)
        result_queue.put({
            'cmd': cmd,
            'returncode': result.returncode,
            'stdout': result.stdout
        })
    
    # Exécuter commandes en parallèle
    result_queue = Queue()
    processes = []
    
    for cmd in [['ls'], ['pwd'], ['date']]:
        p = Process(target=run_in_process, args=(cmd, result_queue))
        p.start()
        processes.append(p)
    
    # Récupérer résultats
    for p in processes:
        p.join()
    
    while not result_queue.empty():
        result = result_queue.get()
        print(f"{result['cmd']}: {result['stdout']}")


[OK] ASYNC ET ASYNCIO


[OK] ASYNCIO SUBPROCESS
    import asyncio
    
    async def run_async(cmd):
        """Exécute commande de façon asynchrone"""
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        
        stdout, stderr = await process.communicate()
        
        return {
            'returncode': process.returncode,
            'stdout': stdout.decode(),
            'stderr': stderr.decode()
        }
    
    # Exécuter
    result = asyncio.run(run_async(['ls', '-l']))
    print(result['stdout'])

[OK] MULTIPLES COMMANDES ASYNC
    async def run_multiple_async(commands):
        """Exécute plusieurs commandes en parallèle (async)"""
        tasks = [run_async(cmd) for cmd in commands]
        results = await asyncio.gather(*tasks)
        return results
    
    commands = [
        ['ls', '-l'],
        ['pwd'],
        ['date'],
        ['whoami']
    ]
    
    results = asyncio.run(run_multiple_async(commands))
    for i, result in enumerate(results):
        print(f"Commande {i}: {result['stdout'].strip()}")

[OK] ASYNC AVEC TIMEOUT
    async def run_with_timeout(cmd, timeout_seconds):
        """Exécute avec timeout asynchrone"""
        try:
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=timeout_seconds
            )
            
            return stdout.decode()
            
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            raise TimeoutError(f"Commande timeout après {timeout_seconds}s")
    
    try:
        result = asyncio.run(run_with_timeout(['sleep', '10'], 5))
    except TimeoutError as e:
        print(e)

[OK] STREAMING ASYNC
    async def stream_output(cmd):
        """Stream output ligne par ligne (async)"""
        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        
        async for line in process.stdout:
            print(f"Output: {line.decode().strip()}")
        
        await process.wait()
        return process.returncode
    
    asyncio.run(stream_output(['ls', '-lR', '/']))


[OK] PERFORMANCE ET OPTIMISATION


[OK] RÉUTILISER PROCESSUS (POOL)
    from multiprocessing import Pool
    
    def run_command(cmd):
        """Fonction pour pool"""
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True)
        return result.stdout
    
    commands = [['ls'], ['pwd'], ['date']] * 10
    
    # Exécuter avec pool
    with Pool(processes=4) as pool:
        results = pool.map(run_command, commands)
    
    print(f"Exécuté {len(results)} commandes")

[OK] CACHING DE RÉSULTATS
    from functools import lru_cache
    import hashlib
    
    @lru_cache(maxsize=128)
    def run_cached(cmd_tuple):
        """Exécute commande avec cache"""
        cmd = list(cmd_tuple)
        result = subprocess.run(cmd,
                               capture_output=True,
                               text=True)
        return result.stdout
    
    # Utilisation (cmd doit être tuple pour être hashable)
    output1 = run_cached(('ls', '-l'))
    output2 = run_cached(('ls', '-l'))  # Utilise cache
    
    print(run_cached.cache_info())

[OK] BATCH PROCESSING
    def batch_process(commands, batch_size=10):
        """Traite commandes par lots"""
        results = []
        
        for i in range(0, len(commands), batch_size):
            batch = commands[i:i+batch_size]
            
            batch_results = []
            for cmd in batch:
                result = subprocess.run(cmd,
                                       capture_output=True,
                                       text=True)
                batch_results.append(result)
            
            results.extend(batch_results)
            
            # Pause entre batches
            if i + batch_size < len(commands):
                time.sleep(0.5)
        
        return results
    
    commands = [['echo', str(i)] for i in range(100)]
    results = batch_process(commands, batch_size=10)

[OK] LIMITER PROCESSUS CONCURRENTS
    from threading import Semaphore
    import threading
    
    class ProcessPool:
        """Pool simple avec limite de concurrence"""
        def __init__(self, max_workers=5):
            self.semaphore = Semaphore(max_workers)
            self.results = []
        
        def run(self, cmd):
            """Exécute commande avec limite"""
            with self.semaphore:
                result = subprocess.run(cmd,
                                       capture_output=True,
                                       text=True)
                self.results.append(result)
    
    pool = ProcessPool(max_workers=3)
    threads = []
    
    commands = [['echo', str(i)] for i in range(20)]
    for cmd in commands:
        t = threading.Thread(target=pool.run, args=(cmd,))
        t.start()
        threads.append(t)
    
    for t in threads:
        t.join()
    
    print(f"Exécuté {len(pool.results)} commandes")


[OK] TESTING SUBPROCESS


[OK] MOCK SUBPROCESS
    from unittest.mock import patch, MagicMock
    
    def my_function():
        """Fonction qui utilise subprocess"""
        result = subprocess.run(['ls', '-l'],
                               capture_output=True,
                               text=True)
        return result.stdout
    
    # Test avec mock
    def test_my_function():
        mock_result = MagicMock()
        mock_result.stdout = "mocked output"
        mock_result.returncode = 0
        
        with patch('subprocess.run', return_value=mock_result) as mock_run:
            output = my_function()
            
            assert output == "mocked output"
            mock_run.assert_called_once_with(
                ['ls', '-l'],
                capture_output=True,
                text=True
            )
    
    test_my_function()

[OK] TESTER ERREURS
    def test_command_error():
        """Test gestion d'erreur"""
        mock_result = MagicMock()
        mock_result.returncode = 1
        mock_result.stderr = "Error message"
        
        with patch('subprocess.run', return_value=mock_result):
            result = subprocess.run(['invalid_command'],
                                   capture_output=True)
            assert result.returncode != 0

[OK] TESTER AVEC VRAIS PROCESSUS
    import tempfile
    
    def test_real_subprocess():
        """Test avec vrai processus"""
        with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
            f.write('test content\n')
            temp_file = f.name
        
        try:
            # Test cat
            result = subprocess.run(['cat', temp_file],
                                   capture_output=True,
                                   text=True)
            assert result.returncode == 0
            assert 'test content' in result.stdout
            
            # Test grep
            result = subprocess.run(['grep', 'test', temp_file],
                                   capture_output=True,
                                   text=True)
            assert result.returncode == 0
            
        finally:
            os.unlink(temp_file)
    
    test_real_subprocess()


[OK] SUBPROCESS ALTERNATIVES


[OK] OS.SYSTEM (OBSOLÈTE)
    # [X] Ne pas utiliser (obsolète et non sécurisé)
    os.system('ls -l')
    
    # [OK] Utiliser subprocess.run à la place
    subprocess.run(['ls', '-l'])

[OK] OS.POPEN (OBSOLÈTE)
    # [X] Ne pas utiliser (obsolète)
    stream = os.popen('ls -l')
    output = stream.read()
    
    # [OK] Utiliser subprocess.run
    result = subprocess.run(['ls', '-l'],
                           capture_output=True,
                           text=True)
    output = result.stdout

[OK] COMMANDS MODULE (OBSOLÈTE, UNIX SEULEMENT)
    # [X] Ne pas utiliser (obsolète, Python 2)
    import commands
    status, output = commands.getstatusoutput('ls -l')
    
    # [OK] Utiliser subprocess
    result = subprocess.run(['ls', '-l'],
                           capture_output=True,
                           text=True)
    status = result.returncode
    output = result.stdout

[OK] SH LIBRARY (ALTERNATIVE EXTERNE)
    # Alternative: bibliothèque 'sh' (pip install sh)
    # import sh
    # output = sh.ls('-l')
    
    # Mais subprocess est préférable (standard library)


[OK] TROUBLESHOOTING COURANT


[OK] "COMMAND NOT FOUND"
    # Problème: commande non trouvée
    # Solution 1: Vérifier PATH
    import shutil
    if shutil.which('mycommand') is None:
        print("Commande non disponible")
    
    # Solution 2: Chemin absolu
    subprocess.run(['/usr/local/bin/mycommand'])
    
    # Solution 3: Ajouter au PATH
    env = os.environ.copy()
    env['PATH'] = '/custom/path:' + env['PATH']
    subprocess.run(['mycommand'], env=env)

[OK] ENCODING ERRORS
    # Problème: erreur d'encodage
    # Solution: Spécifier encodage et gestion erreurs
    result = subprocess.run(['command'],
                           capture_output=True,
                           text=True,
                           encoding='utf-8',
                           errors='replace')  # ou 'ignore'

[OK] PROCESSUS ZOMBIE
    # Problème: processus zombie
    # Solution: Toujours appeler wait()
    process = subprocess.Popen(['command'])
    # ... faire autre chose ...
    process.wait()  # Important!

[OK] BUFFER OVERFLOW
    # Problème: deadlock sur gros output
    # [X] Mauvais
    process = subprocess.Popen(['command'],
                              stdout=subprocess.PIPE)
    # Commande génère beaucoup de données
    process.wait()  # Deadlock si buffer plein!
    
    # [OK] Bon
    process = subprocess.Popen(['command'],
                              stdout=subprocess.PIPE)
    stdout, stderr = process.communicate()  # Gère buffer

[OK] PERMISSION DENIED
    # Problème: permission refusée
    # Solution 1: Vérifier permissions fichier
    os.chmod('script.sh', 0o755)
    subprocess.run(['./script.sh'])
    
    # Solution 2: Utiliser interpréteur
    subprocess.run(['bash', 'script.sh'])

[OK] TIMEOUT NON RESPECTÉ
    # Problème: timeout ne fonctionne pas
    # Cause: processus enfants non tués
    # Solution: Utiliser process groups (Unix)
    import signal
    
    process = subprocess.Popen(['command'],
                              preexec_fn=os.setsid)
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        os.killpg(os.getpgid(process.pid), signal.SIGTERM)


[OK] RÉCAPITULATIF - QUAND UTILISER QUOI


[OK] DÉCISION RAPIDE
    """
    subprocess.run() 
    -> Cas général, commandes simples
    -> Python 3.5+
    -> Recommandé par défaut
    
    subprocess.Popen()
    -> Contrôle avancé
    -> Communication bidirectionnelle
    -> Processus en arrière-plan
    -> Streaming de données
    
    subprocess.call/check_call/check_output()
    -> Code legacy (Python < 3.5)
    -> Migration vers run() recommandée
    
    asyncio.create_subprocess_exec()
    -> Applications async
    -> Multiple commandes parallèles
    -> Event loop asyncio
    """

[OK] PATTERNS RECOMMANDÉS
    # [OK] Simple et sûr
    subprocess.run(['command', 'arg'], check=True)
    
    # [OK] Capturer output
    result = subprocess.run(['command'],
                           capture_output=True,
                           text=True,
                           check=True)
    
    # [OK] Avec timeout
    subprocess.run(['command'],
                  timeout=10,
                  check=True)
    
    # [OK] Input depuis string
    subprocess.run(['command'],
                  input='data',
                  text=True,
                  check=True)
    
    # [OK] Processus long avec streaming
    with subprocess.Popen(['command'],
                         stdout=subprocess.PIPE,
                         text=True) as proc:
        for line in proc.stdout:
            process_line(line)

[OK] ANTIPATTERNS À ÉVITER
    # [X] shell=True avec input utilisateur
    subprocess.run(f'command {user_input}', shell=True)
    
    # [X] Pas de gestion d'erreur
    subprocess.run(['command'])  # Peut échouer silencieusement
    
    # [X] Pas de timeout
    subprocess.run(['long_command'])  # Peut bloquer indéfiniment
    
    # [X] Lire stdout sans PIPE
    p = subprocess.Popen(['command'])
    p.stdout.read()  # Erreur!
    
    # [X] Ignorer processus zombie
    subprocess.Popen(['command'])  # Sans wait()


[OK] RESSOURCES ET RÉFÉRENCES


[OK] DOCUMENTATION OFFICIELLE
    # Python subprocess documentation:
    # https://docs.python.org/3/library/subprocess.html
    
    # PEP 324 - subprocess module:
    # https://www.python.org/dev/peps/pep-0324/

[OK] MODULES COMPLÉMENTAIRES
    # sh - Python subprocess alternative
    # pip install sh
    
    # plumbum - Shell combinators library
    # pip install plumbum
    
    # invoke - Task execution tool
    # pip install invoke

[OK] COMMANDES UTILES POUR DEBUGGING
    # Afficher version Python
    subprocess.run([sys.executable, '--version'])
    
    # Afficher toutes les variables d'environnement
    subprocess.run(['printenv'])
    
    # Tester PATH
    subprocess.run(['which', 'command'])
    
    # Afficher PID du processus
    process = subprocess.Popen(['sleep', '100'])
    print(f"PID: {process.pid}")


# FIN DU CHEATSHEET SUBPROCESS