
# Fichier: python_cheats/cheatsheets/asyncio.txt


import asyncio
import time
from typing import Any, List, Coroutine


[OK] 1. CONCEPTS DE BASE


# COROUTINE: Fonction définie avec async def
async def my_coroutine():
    """Fonction asynchrone qui peut être suspendue"""
    await asyncio.sleep(1)  # Suspend l'exécution sans bloquer
    return "Result"

# AWAIT: Attend la complétion d'une coroutine
async def example_await():
    """Démontre l'utilisation de await"""
    result = await my_coroutine()  # Pause ici jusqu'à complétion
    print(result)

# TASK: Enveloppe une coroutine pour exécution concurrente
async def create_tasks():
    """Les tasks permettent l'exécution en arrière-plan"""
    task1 = asyncio.create_task(my_coroutine())  # Lance immédiatement
    task2 = asyncio.create_task(my_coroutine())  # Lance immédiatement
    
    # Les deux s'exécutent en parallèle
    result1 = await task1
    result2 = await task2



[OK] 2. EXÉCUTION DE COROUTINES


# RUN: Point d'entrée principal (Python 3.7+)
def run_simple():
    """Méthode recommandée pour lancer asyncio"""
    asyncio.run(my_coroutine())

# RUN avec paramètres (Python 3.10+)
def run_with_debug():
    """Options de debug et configuration"""
    asyncio.run(my_coroutine(), debug=True)  # Active mode debug

# EVENT LOOP manuel (ancien style, rarement nécessaire)
def run_old_style():
    """Méthode historique, éviter si possible"""
    loop = asyncio.get_event_loop()
    try:
        result = loop.run_until_complete(my_coroutine())
    finally:
        loop.close()

# GET_RUNNING_LOOP: Obtenir loop actif
async def get_current_loop():
    """Accès au loop dans contexte async"""
    loop = asyncio.get_running_loop()  # Seulement dans async context
    print(f"Loop: {loop}")

# GET_EVENT_LOOP: Obtenir ou créer loop (deprecated)
def get_or_create_loop():
    """Ancien style - utiliser asyncio.run() à la place"""
    loop = asyncio.get_event_loop()  # Peut créer nouveau loop
    return loop



[OK] 3. CRÉATION ET GESTION DE TASKS


# CREATE_TASK: Créer task depuis coroutine
async def create_task_example():
    """Crée task qui s'exécute immédiatement en arrière-plan"""
    task = asyncio.create_task(my_coroutine())
    # La coroutine commence immédiatement
    result = await task  # Attend la complétion

# CREATE_TASK avec nom (Python 3.8+)
async def named_task():
    """Nomme les tasks pour debugging"""
    task = asyncio.create_task(my_coroutine(), name="MyTask")
    print(f"Task name: {task.get_name()}")

# ENSURE_FUTURE: Alternative à create_task
async def ensure_future_example():
    """Transforme coroutine/future en task"""
    task = asyncio.ensure_future(my_coroutine())
    result = await task

# CURRENT_TASK: Obtenir task actuelle
async def get_current_task():
    """Référence à la task en cours d'exécution"""
    task = asyncio.current_task()
    print(f"Current: {task.get_name()}")

# ALL_TASKS: Lister toutes les tasks
async def list_all_tasks():
    """Obtenir ensemble de toutes les tasks"""
    tasks = asyncio.all_tasks()
    for task in tasks:
        print(f"- {task.get_name()}: {task.done()}")

# TASK CANCELLATION: Annuler une task
async def cancel_task_example():
    """Annulation gracieuse de tasks"""
    task = asyncio.create_task(asyncio.sleep(10))
    
    await asyncio.sleep(1)
    task.cancel()  # Demande d'annulation
    
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

# TASK RESULT: Obtenir résultat
async def get_task_result():
    """Récupérer résultat d'une task complétée"""
    task = asyncio.create_task(my_coroutine())
    await task
    
    if task.done():
        result = task.result()  # Lève exception si task a échoué
        print(result)

# TASK EXCEPTION: Gérer exceptions
async def handle_task_exception():
    """Vérifier si task a levé exception"""
    async def failing():
        raise ValueError("Error!")
    
    task = asyncio.create_task(failing())
    
    try:
        await task
    except ValueError:
        pass
    
    if task.done():
        exc = task.exception()  # None si pas d'exception
        if exc:
            print(f"Task failed with: {exc}")



[OK] 4. EXÉCUTION CONCURRENTE


# GATHER: Exécuter plusieurs coroutines en parallèle
async def gather_example():
    """Attend toutes les coroutines, retourne résultats dans l'ordre"""
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    # results = [result1, result2, result3]
    print(results)

# GATHER avec gestion d'erreurs
async def gather_with_errors():
    """Par défaut, première erreur annule tout"""
    try:
        results = await asyncio.gather(
            good_task(),
            failing_task(),
            return_exceptions=False  # Lève exception
        )
    except Exception as e:
        print(f"Error: {e}")

# GATHER retournant exceptions
async def gather_return_exceptions():
    """Continue même si erreurs, retourne exceptions"""
    results = await asyncio.gather(
        good_task(),
        failing_task(),
        return_exceptions=True  # Exceptions dans results
    )
    # results peut contenir mix de résultats et exceptions
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

# WAIT: Plus de contrôle que gather
async def wait_example():
    """Attendre tasks avec différentes conditions"""
    tasks = [
        asyncio.create_task(fetch_data(i))
        for i in range(5)
    ]
    
    # Attendre toutes les tasks
    done, pending = await asyncio.wait(tasks)
    # done: set de tasks complétées
    # pending: set de tasks en cours (vide ici)

# WAIT avec FIRST_COMPLETED
async def wait_first():
    """Retourne dès qu'une task se termine"""
    tasks = [
        asyncio.create_task(fetch_data(i))
        for i in range(5)
    ]
    
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    
    # done contient 1 task, pending en contient 4
    # Annuler les tasks restantes
    for task in pending:
        task.cancel()

# WAIT avec FIRST_EXCEPTION
async def wait_first_exception():
    """Retourne dès qu'une task échoue ou toutes réussissent"""
    tasks = [asyncio.create_task(t()) for t in [good_task, failing_task]]
    
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_EXCEPTION
    )

# WAIT avec timeout
async def wait_timeout():
    """Limite temps d'attente"""
    tasks = [asyncio.create_task(slow_task()) for _ in range(3)]
    
    done, pending = await asyncio.wait(
        tasks,
        timeout=5.0  # Secondes
    )
    
    print(f"Completed: {len(done)}, Pending: {len(pending)}")

# AS_COMPLETED: Traiter résultats au fur et à mesure
async def as_completed_example():
    """Itère sur coroutines dans l'ordre de complétion"""
    coros = [fetch_data(i) for i in range(5)]
    
    for coro in asyncio.as_completed(coros):
        result = await coro  # Premier terminé en premier
        print(f"Got: {result}")

# AS_COMPLETED avec timeout
async def as_completed_timeout():
    """Timeout global pour toutes les coroutines"""
    coros = [fetch_data(i) for i in range(5)]
    
    for coro in asyncio.as_completed(coros, timeout=10):
        try:
            result = await coro
            print(result)
        except asyncio.TimeoutError:
            print("Some tasks timed out")
            break



[OK] 5. TIMEOUTS ET DÉLAIS


# SLEEP: Pause non-bloquante
async def sleep_example():
    """Suspend coroutine sans bloquer event loop"""
    print("Starting")
    await asyncio.sleep(1.5)  # Pause 1.5 secondes
    print("After sleep")

# WAIT_FOR: Timeout pour une coroutine
async def wait_for_example():
    """Limite temps d'exécution d'une coroutine"""
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=5.0
        )
    except asyncio.TimeoutError:
        print("Operation timed out!")
        result = None

# WAIT_FOR annule la coroutine en timeout
async def wait_for_cancels():
    """La coroutine est annulée si timeout"""
    async def long_task():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("Task was cancelled due to timeout")
            raise
    
    try:
        await asyncio.wait_for(long_task(), timeout=1.0)
    except asyncio.TimeoutError:
        print("Timed out")

# TIMEOUT: Context manager pour timeout (Python 3.11+)
async def timeout_context():
    """Alternative moderne à wait_for"""
    async with asyncio.timeout(5.0):
        await slow_operation()
    # TimeoutError levée si dépasse 5 secondes

# TIMEOUT_AT: Timeout à timestamp absolu (Python 3.11+)
async def timeout_at_example():
    """Timeout basé sur temps absolu"""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 5.0  # Dans 5 secondes
    
    async with asyncio.timeout_at(deadline):
        await slow_operation()



[OK] 6. SYNCHRONISATION PRIMITIVES


# LOCK: Verrou pour section critique
lock = asyncio.Lock()

async def lock_example():
    """Protège section critique d'accès concurrent"""
    async with lock:
        # Une seule coroutine à la fois ici
        await asyncio.sleep(1)
        print("Critical section")

# LOCK manuel
async def lock_manual():
    """Utilisation explicite acquire/release"""
    await lock.acquire()
    try:
        await critical_operation()
    finally:
        lock.release()

# LOCK locked/locked check
async def lock_check():
    """Vérifier état du lock"""
    if lock.locked():
        print("Lock is held")
    else:
        async with lock:
            print("Acquired lock")

# EVENT: Signal entre coroutines
event = asyncio.Event()

async def event_waiter():
    """Attend signal d'un event"""
    print("Waiting for event...")
    await event.wait()  # Bloque jusqu'à set()
    print("Event received!")

async def event_setter():
    """Envoie signal"""
    await asyncio.sleep(2)
    event.set()  # Débloque tous les waiters

async def event_example():
    """Coordination entre coroutines"""
    await asyncio.gather(
        event_waiter(),
        event_setter()
    )

# EVENT clear/is_set
async def event_operations():
    """Opérations sur events"""
    event.set()        # Active l'event
    assert event.is_set()  # Vérifie état
    event.clear()      # Désactive l'event
    assert not event.is_set()

# CONDITION: Lock + Event combinés
condition = asyncio.Condition()

async def condition_consumer():
    """Attend condition spécifique"""
    async with condition:
        await condition.wait()  # Attend notification
        print("Condition met")

async def condition_producer():
    """Notifie condition"""
    await asyncio.sleep(1)
    async with condition:
        condition.notify()  # Notifie 1 waiter
        # condition.notify_all()  # Notifie tous

# SEMAPHORE: Limite nombre de coroutines concurrentes
semaphore = asyncio.Semaphore(3)  # Max 3 simultanées

async def semaphore_example():
    """Limite concurrence à 3"""
    async with semaphore:
        # Max 3 coroutines ici simultanément
        await perform_operation()

# BOUNDEDSEMAPHORE: Semaphore avec vérification release
bounded_sem = asyncio.BoundedSemaphore(3)

async def bounded_semaphore():
    """Empêche release() excessifs"""
    async with bounded_sem:
        await operation()
    # bounded_sem.release()  # ValueError si déjà à max

# BARRIER: Synchronisation de N coroutines (Python 3.11+)
barrier = asyncio.Barrier(3)  # Attend 3 coroutines

async def barrier_example(n):
    """Toutes attendent que 3 arrivent au barrier"""
    print(f"Coroutine {n} arrived")
    await barrier.wait()  # Bloque jusqu'à 3 arrivent
    print(f"Coroutine {n} continues")



[OK] 7. QUEUES


# QUEUE: File FIFO asynchrone
async def queue_example():
    """Producer-consumer avec queue"""
    queue = asyncio.Queue(maxsize=10)  # Limite taille
    
    async def producer():
        for i in range(5):
            await queue.put(i)  # Ajoute élément
            print(f"Produced: {i}")
        await queue.put(None)  # Signal fin
    
    async def consumer():
        while True:
            item = await queue.get()  # Récupère élément
            if item is None:
                break
            print(f"Consumed: {item}")
            queue.task_done()  # Marque comme traité
    
    await asyncio.gather(producer(), consumer())

# QUEUE join
async def queue_join():
    """Attendre traitement complet de tous les items"""
    queue = asyncio.Queue()
    
    async def worker():
        while True:
            item = await queue.get()
            await process(item)
            queue.task_done()
    
    # Remplir queue
    for i in range(10):
        await queue.put(i)
    
    # Lancer workers
    workers = [asyncio.create_task(worker()) for _ in range(3)]
    
    # Attendre traitement complet
    await queue.join()
    
    # Arrêter workers
    for w in workers:
        w.cancel()

# QUEUE méthodes non-bloquantes
async def queue_nonblocking():
    """put_nowait et get_nowait pour opérations immédiates"""
    queue = asyncio.Queue()
    
    try:
        queue.put_nowait(1)  # Lève QueueFull si pleine
        item = queue.get_nowait()  # Lève QueueEmpty si vide
    except asyncio.QueueFull:
        print("Queue is full")
    except asyncio.QueueEmpty:
        print("Queue is empty")

# QUEUE informations
async def queue_info():
    """Vérifier état de la queue"""
    queue = asyncio.Queue(maxsize=5)
    
    print(f"Size: {queue.qsize()}")
    print(f"Empty: {queue.empty()}")
    print(f"Full: {queue.full()}")
    print(f"Max size: {queue.maxsize}")

# PRIORITYQUEUE: Queue avec priorités
async def priority_queue_example():
    """Items traités par ordre de priorité"""
    queue = asyncio.PriorityQueue()
    
    await queue.put((3, "low priority"))
    await queue.put((1, "high priority"))
    await queue.put((2, "medium priority"))
    
    # Sort dans l'ordre: high, medium, low
    while not queue.empty():
        priority, item = await queue.get()
        print(f"{priority}: {item}")

# LIFOQUEUE: Last-In-First-Out (stack)
async def lifo_queue_example():
    """Stack asynchrone"""
    queue = asyncio.LifoQueue()
    
    await queue.put(1)
    await queue.put(2)
    await queue.put(3)
    
    # Sort dans l'ordre: 3, 2, 1
    while not queue.empty():
        item = await queue.get()
        print(item)



[OK] 8. STREAMS (I/O RÉSEAU)


# SERVEUR TCP
async def tcp_server():
    """Crée serveur TCP asynchrone"""
    async def handle_client(reader, writer):
        """Traite connexion client"""
        data = await reader.read(100)  # Lit 100 bytes max
        message = data.decode()
        
        addr = writer.get_extra_info('peername')
        print(f"Received {message} from {addr}")
        
        # Réponse
        writer.write(b"Hello from server")
        await writer.drain()  # Flush buffer
        
        writer.close()
        await writer.wait_closed()
    
    server = await asyncio.start_server(
        handle_client,
        '127.0.0.1',
        8888
    )
    
    async with server:
        await server.serve_forever()

# CLIENT TCP
async def tcp_client():
    """Connexion client TCP"""
    reader, writer = await asyncio.open_connection(
        '127.0.0.1',
        8888
    )
    
    # Envoi
    writer.write(b"Hello server")
    await writer.drain()
    
    # Réception
    data = await reader.read(100)
    print(f"Received: {data.decode()}")
    
    writer.close()
    await writer.wait_closed()

# READLINE: Lecture ligne par ligne
async def read_lines():
    """Lit données ligne par ligne"""
    reader, writer = await asyncio.open_connection('example.com', 80)
    
    while True:
        line = await reader.readline()
        if not line:
            break
        print(line.decode().strip())
    
    writer.close()

# READEXACTLY: Lire nombre exact de bytes
async def read_exact():
    """Garantit lecture de N bytes"""
    reader, writer = await asyncio.open_connection('host', 8888)
    
    # Lit exactement 10 bytes ou lève EOFError
    data = await reader.readexactly(10)

# READUNTIL: Lire jusqu'à séparateur
async def read_until():
    """Lit jusqu'à trouver séparateur"""
    reader, writer = await asyncio.open_connection('host', 8888)
    
    # Lit jusqu'à \n
    data = await reader.readuntil(b'\n')

# SERVEUR UNIX SOCKET
async def unix_socket_server():
    """Serveur Unix socket (Linux/Mac uniquement)"""
    async def handle(reader, writer):
        data = await reader.read(100)
        writer.write(data)
        await writer.drain()
        writer.close()
    
    server = await asyncio.start_unix_server(
        handle,
        '/tmp/mysocket'
    )
    
    async with server:
        await server.serve_forever()



[OK] 9. SUBPROCESSES


# CREATE_SUBPROCESS_EXEC: Lancer commande
async def run_command():
    """Exécute commande système de manière asynchrone"""
    process = await asyncio.create_subprocess_exec(
        'ls', '-la',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    stdout, stderr = await process.communicate()
    
    print(f"Return code: {process.returncode}")
    print(f"Output: {stdout.decode()}")

# CREATE_SUBPROCESS_SHELL: Avec shell
async def run_shell():
    """Exécute via shell (attention sécurité!)"""
    process = await asyncio.create_subprocess_shell(
        'ls -la | grep py',
        stdout=asyncio.subprocess.PIPE
    )
    
    stdout, _ = await process.communicate()
    print(stdout.decode())

# Communication avec subprocess
async def communicate_subprocess():
    """Envoie données et récupère output"""
    process = await asyncio.create_subprocess_exec(
        'cat',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )
    
    stdout, _ = await process.communicate(b'Hello subprocess')
    print(stdout.decode())

# Lecture stream subprocess
async def read_subprocess_stream():
    """Lit output ligne par ligne"""
    process = await asyncio.create_subprocess_exec(
        'python', 'script.py',
        stdout=asyncio.subprocess.PIPE
    )
    
    async for line in process.stdout:
        print(line.decode().strip())
    
    await process.wait()

# Terminer subprocess
async def terminate_subprocess():
    """Arrêt gracieux ou forcé"""
    process = await asyncio.create_subprocess_exec('sleep', '100')
    
    process.terminate()  # SIGTERM
    # process.kill()     # SIGKILL
    
    await process.wait()



[OK] 10. FUTURES ET CALLBACKS


# FUTURE: Conteneur pour résultat futur
async def future_example():
    """Future représente valeur qui arrivera plus tard"""
    future = asyncio.Future()
    
    async def set_result():
        await asyncio.sleep(1)
        future.set_result("Result!")  # Résout le future
    
    asyncio.create_task(set_result())
    
    result = await future  # Attend résolution
    print(result)

# FUTURE avec exception
async def future_exception():
    """Future peut contenir exception"""
    future = asyncio.Future()
    
    future.set_exception(ValueError("Error!"))
    
    try:
        await future
    except ValueError as e:
        print(f"Caught: {e}")

# FUTURE callbacks
async def future_callbacks():
    """Exécute callback quand future résolu"""
    future = asyncio.Future()
    
    def callback(fut):
        print(f"Future resolved: {fut.result()}")
    
    future.add_done_callback(callback)
    future.set_result(42)

# FUTURE état
async def future_state():
    """Vérifier état du future"""
    future = asyncio.Future()
    
    print(f"Done: {future.done()}")
    print(f"Cancelled: {future.cancelled()}")
    
    future.set_result(1)
    print(f"Done: {future.done()}")
    
    if future.done():
        result = future.result()  # Pas de await nécessaire

# WRAP_FUTURE: Convertir concurrent.futures en asyncio
from concurrent.futures import ThreadPoolExecutor

async def wrap_future_example():
    """Intègre futures threading avec asyncio"""
    executor = ThreadPoolExecutor()
    
    def blocking():
        time.sleep(1)
        return "Done"
    
    loop = asyncio.get_running_loop()
    future = executor.submit(blocking)
    result = await asyncio.wrap_future(future)



[OK] 11. EXÉCUTION DE CODE BLOQUANT


# RUN_IN_EXECUTOR: Code bloquant en thread pool
async def run_in_executor():
    """Exécute fonction bloquante sans bloquer event loop"""
    def blocking_io():
        time.sleep(2)
        return "Completed"
    
    loop = asyncio.get_running_loop()
    # None = utilise executor par défaut (ThreadPoolExecutor)
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

# RUN_IN_EXECUTOR avec ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor

async def run_in_process():
    """CPU-intensive task dans process séparé"""
    def cpu_bound(n):
        return sum(i * i for i in range(n))
    
    loop = asyncio.get_running_loop()
    executor = ProcessPoolExecutor()
    
    result = await loop.run_in_executor(executor, cpu_bound, 10_000_000)
    print(result)

# RUN_IN_EXECUTOR avec arguments
async def executor_with_args():
    """Passer arguments à fonction"""
    from functools import partial
    
    def func(a, b, c):
        return a + b + c
    
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(
        None,
        partial(func, 1, 2, c=3)
    )

# TO_THREAD: Simplification Python 3.9+
async def to_thread_example():
    """Alternative moderne à run_in_executor"""
    def blocking():
        time.sleep(1)
        return "Done"
    
    result = await asyncio.to_thread(blocking)



[OK] 12. CONTEXT MANAGERS ET ITERATORS ASYNC


# ASYNC CONTEXT MANAGER
class AsyncResource:
    """Ressource avec acquisition/libération async"""
    
    async def __aenter__(self):
        """Appelé à l'entrée du with"""
        print("Acquiring resource")
        await asyncio.sleep(0.1)
        self.resource = "Resource"
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Appelé à la sortie du with"""
        print("Releasing resource")
        await asyncio.sleep(0.1)
        return False  # Propage exceptions

async def use_async_context():
    """Utilisation d'async context manager"""
    async with AsyncResource() as resource:
        print(f"Using {resource.resource}")

# ASYNC ITERATOR
class AsyncCounter:
    """Iterator qui génère valeurs de façon asynchrone"""
    
    def __init__(self, max_value):
        self.max = max_value
        self.current = 0
    
    def __aiter__(self):
        """Retourne l'iterator lui-même"""
        return self
    
    async def __anext__(self):
        """Génère prochaine valeur"""
        if self.current >= self.max:
            raise StopAsyncIteration
        
        await asyncio.sleep(0.1)
        self.current += 1
        return self.current

async def use_async_iterator():
    """Itération asynchrone"""
    async for value in AsyncCounter(5):
        print(value)

# ASYNC GENERATOR
async def async_generator(n):
    """Generator asynchrone avec yield"""
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def use_async_generator():
    """Consomme async generator"""
    async for value in async_generator(5):
        print(value)

# ASYNC GENERATOR expression
async def async_gen_expr():
    """Compréhension avec async for"""
    gen = (i async for i in async_generator(5))
    
    async for value in gen:
        print(value)



[OK] 13. GESTION D'ERREURS ET EXCEPTIONS


# CANCELLED ERROR: Task annulée
async def handle_cancellation():
    """Gestion propre de l'annulation"""
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("Task cancelled, cleaning up...")
        # Cleanup code
        raise  # Re-raise pour propager

# TIMEOUT ERROR
async def handle_timeout():
    """Gestion des timeouts"""
    try:
        await asyncio.wait_for(slow_operation(), timeout=5)
    except asyncio.TimeoutError:
        print("Operation timed out")

# GATHER avec exceptions
async def gather_error_handling():
    """Récupérer toutes les exceptions"""
    async def task(n):
        if n == 2:
            raise ValueError(f"Error in task {n}")
        return n
    
    results = await asyncio.gather(
        task(1),
        task(2),
        task(3),
        return_exceptions=True
    )
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} result: {result}")

# SHIELD: Protéger contre annulation
async def shield_example():
    """Empêche annulation d'opération critique"""
    async def critical():
        await asyncio.sleep(5)
        return "Important result"
    
    try:
        # Shield protège critical() même si annulation
        result = await asyncio.shield(critical())
    except asyncio.CancelledError:
        print("Outer cancelled, but critical() continues")
        raise



[OK] 14. DEBUGGING ET MONITORING


# DEBUG MODE
def enable_debug():
    """Active mode debug pour asyncio"""
    asyncio.run(main(), debug=True)
    # Affiche warnings pour coroutines non awaitées, etc.

# SLOW CALLBACK DURATION
async def set_slow_callback_duration():
    """Alerte si callback prend trop de temps"""
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.1  # Warning si callback > 100ms

# CURRENT_TASK pour debugging
async def debug_current_task():
    """Informations sur task actuelle"""
    task = asyncio.current_task()
    print(f"Task: {task.get_name()}")
    print(f"Coro: {task.get_coro()}")
    print(f"Stack: {task.get_stack()}")

# ALL_TASKS monitoring
async def monitor_tasks():
    """Liste et surveille toutes les tasks"""
    tasks = asyncio.all_tasks()
    for task in tasks:
        print(f"- {task.get_name()}: done={task.done()}, cancelled={task.cancelled()}")

# EXCEPTION HANDLER personnalisé
def custom_exception_handler(loop, context):
    """Gestionnaire global d'exceptions"""
    print(f"Exception caught: {context['message']}")
    exception = context.get('exception')
    if exception:
        print(f"Exception type: {type(exception)}")

async def set_exception_handler():
    """Configure handler d'exceptions"""
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(custom_exception_handler)

# CALL_EXCEPTION_HANDLER manuel
async def call_exception_handler():
    """Déclenche handler manuellement"""
    loop = asyncio.get_running_loop()
    loop.call_exception_handler({
        'message': 'Custom error',
        'exception': ValueError('Test')
    })



[OK] 15. SCHEDULING ET TIMING


# CALL_SOON: Planifier callback dès que possible
async def call_soon_example():
    """Exécute callback au prochain tour du loop"""
    loop = asyncio.get_running_loop()
    
    def callback():
        print("Called soon")
    
    loop.call_soon(callback)
    await asyncio.sleep(0)  # Laisse loop exécuter callback

# CALL_SOON_THREADSAFE: Depuis autre thread
def call_from_thread():
    """Planifier depuis thread non-asyncio"""
    loop = asyncio.get_running_loop()
    
    def callback():
        print("Called from thread")
    
    # Thread-safe
    loop.call_soon_threadsafe(callback)

# CALL_LATER: Planifier après délai
async def call_later_example():
    """Exécute callback après délai"""
    loop = asyncio.get_running_loop()
    
    def callback():
        print("Called after 2 seconds")
    
    loop.call_later(2.0, callback)  # Délai en secondes
    await asyncio.sleep(3)  # Attend exécution

# CALL_AT: Planifier à timestamp absolu
async def call_at_example():
    """Exécute à moment précis"""
    loop = asyncio.get_running_loop()
    
    # Dans 5 secondes
    when = loop.time() + 5.0
    
    def callback():
        print("Called at specific time")
    
    loop.call_at(when, callback)

# HANDLE pour annuler callbacks
async def cancel_callback():
    """Annule callback planifié"""
    loop = asyncio.get_running_loop()
    
    handle = loop.call_later(10.0, lambda: print("This won't run"))
    
    # Annulation
    handle.cancel()
    print(f"Cancelled: {handle.cancelled()}")

# TIME: Horloge du loop
async def loop_time():
    """Horloge monotone du loop"""
    loop = asyncio.get_running_loop()
    
    start = loop.time()
    await asyncio.sleep(1)
    end = loop.time()
    
    print(f"Elapsed: {end - start:.2f}s")



[OK] 16. POLITIQUES ET CONFIGURATION


# EVENT LOOP POLICY
def get_event_loop_policy():
    """Obtenir politique du loop"""
    policy = asyncio.get_event_loop_policy()
    print(f"Policy: {policy}")

# SET EVENT LOOP POLICY
def set_custom_policy():
    """Configurer politique personnalisée"""
    # Sur Windows, pour subprocess
    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(
            asyncio.WindowsProactorEventLoopPolicy()
        )

# NEW EVENT LOOP
def create_new_loop():
    """Créer nouveau loop manuellement"""
    new_loop = asyncio.new_event_loop()
    asyncio.set_event_loop(new_loop)
    try:
        new_loop.run_until_complete(my_coroutine())
    finally:
        new_loop.close()

# DEFAULT EXECUTOR
async def set_default_executor():
    """Configure executor par défaut"""
    from concurrent.futures import ThreadPoolExecutor
    
    loop = asyncio.get_running_loop()
    executor = ThreadPoolExecutor(max_workers=4)
    loop.set_default_executor(executor)



# 17. INTÉGRATION AVEC BIBLIOTHÈQUES


# AIOHTTP: Requêtes HTTP asynchrones
async def aiohttp_example():
    """Client HTTP asynchrone"""
    import aiohttp
    
    async with aiohttp.ClientSession() as session:
        # GET request
        async with session.get('https://api.example.com/data') as response:
            data = await response.json()
            print(data)
        
        # POST request
        async with session.post(
            'https://api.example.com/submit',
            json={'key': 'value'}
        ) as response:
            result = await response.text()

# AIOHTTP avec timeout
async def aiohttp_timeout():
    """Requêtes avec timeout"""
    import aiohttp
    
    timeout = aiohttp.ClientTimeout(total=10)  # 10s total
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://example.com') as response:
            data = await response.read()

# AIOHTTP concurrence
async def fetch_multiple_urls():
    """Fetch plusieurs URLs en parallèle"""
    import aiohttp
    
    urls = [
        'https://example.com',
        'https://example.org',
        'https://example.net'
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            data = await response.text()
            print(f"{response.url}: {len(data)} bytes")

# AIOFILES: Fichiers asynchrones
async def aiofiles_example():
    """I/O fichiers non-bloquant"""
    import aiofiles
    
    # Lecture
    async with aiofiles.open('file.txt', 'r') as f:
        content = await f.read()
        print(content)
    
    # Écriture
    async with aiofiles.open('output.txt', 'w') as f:
        await f.write('Hello async files!')
    
    # Lecture ligne par ligne
    async with aiofiles.open('large_file.txt', 'r') as f:
        async for line in f:
            print(line.strip())

# AIOFILES opérations
async def aiofiles_operations():
    """Opérations sur fichiers"""
    import aiofiles
    import aiofiles.os
    
    # Renommer
    await aiofiles.os.rename('old.txt', 'new.txt')
    
    # Supprimer
    await aiofiles.os.remove('file.txt')
    
    # Créer répertoire
    await aiofiles.os.mkdir('newdir')
    
    # Statistiques
    stat = await aiofiles.os.stat('file.txt')
    print(f"Size: {stat.st_size}")

# AIOMYSQL / AIOPG: Bases de données
async def database_async():
    """Connexion DB asynchrone"""
    import aiomysql
    
    pool = await aiomysql.create_pool(
        host='localhost',
        user='user',
        password='pass',
        db='database'
    )
    
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM users")
            rows = await cur.fetchall()
            print(rows)
    
    pool.close()
    await pool.wait_closed()

# REDIS asynchrone
async def redis_async():
    """Redis avec aioredis"""
    import aioredis
    
    redis = await aioredis.create_redis_pool('redis://localhost')
    
    # Set
    await redis.set('key', 'value')
    
    # Get
    value = await redis.get('key', encoding='utf-8')
    print(value)
    
    redis.close()
    await redis.wait_closed()



[OK] 18. PATTERNS AVANCÉS


# WORKER POOL pattern
async def worker_pool(num_workers=5):
    """Pool de workers consommant queue"""
    queue = asyncio.Queue()
    
    async def worker(name):
        """Worker qui traite items"""
        while True:
            item = await queue.get()
            if item is None:
                break
            
            try:
                await process_item(item)
            except Exception as e:
                print(f"Worker {name} error: {e}")
            finally:
                queue.task_done()
    
    # Créer workers
    workers = [
        asyncio.create_task(worker(f"Worker-{i}"))
        for i in range(num_workers)
    ]
    
    # Ajouter travail
    for i in range(20):
        await queue.put(i)
    
    # Attendre completion
    await queue.join()
    
    # Arrêter workers
    for _ in range(num_workers):
        await queue.put(None)
    
    await asyncio.gather(*workers)

# RETRY pattern
async def retry_with_backoff(coro_func, max_attempts=3, backoff=1.0):
    """Réessaye avec délai exponentiel"""
    for attempt in range(max_attempts):
        try:
            return await coro_func()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            
            wait = backoff * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed, retrying in {wait}s...")
            await asyncio.sleep(wait)

# CIRCUIT BREAKER pattern
class CircuitBreaker:
    """Empêche appels répétés vers service défaillant"""
    
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half-open
    
    async def call(self, coro):
        """Exécute coroutine avec circuit breaker"""
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'half-open'
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = await coro
            if self.state == 'half-open':
                self.state = 'closed'
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            
            if self.failures >= self.failure_threshold:
                self.state = 'open'
            
            raise

# RATE LIMITER pattern
class RateLimiter:
    """Limite taux d'exécution"""
    
    def __init__(self, rate=10, per=1.0):
        """rate requêtes par 'per' secondes"""
        self.rate = rate
        self.per = per
        self.allowance = rate
        self.last_check = time.time()
    
    async def acquire(self):
        """Attend si limite dépassée"""
        current = time.time()
        elapsed = current - self.last_check
        self.last_check = current
        self.allowance += elapsed * (self.rate / self.per)
        
        if self.allowance > self.rate:
            self.allowance = self.rate
        
        if self.allowance < 1.0:
            sleep_time = (1.0 - self.allowance) * (self.per / self.rate)
            await asyncio.sleep(sleep_time)
            self.allowance = 0.0
        else:
            self.allowance -= 1.0

# ASYNC CONTEXT POOL
class AsyncConnectionPool:
    """Pool de connexions asynchrones"""
    
    def __init__(self, create_conn, max_size=10):
        self.create_conn = create_conn
        self.max_size = max_size
        self.pool = asyncio.Queue(maxsize=max_size)
        self.size = 0
    
    async def acquire(self):
        """Obtient connexion du pool"""
        if self.pool.empty() and self.size < self.max_size:
            conn = await self.create_conn()
            self.size += 1
            return conn
        return await self.pool.get()
    
    async def release(self, conn):
        """Remet connexion dans pool"""
        await self.pool.put(conn)
    
    async def __aenter__(self):
        self.conn = await self.acquire()
        return self.conn
    
    async def __aexit__(self, *args):
        await self.release(self.conn)



[OK] 19. TESTING ASYNCIO


# UNITTEST avec asyncio
import unittest

class TestAsync(unittest.TestCase):
    """Tests pour code asynchrone"""
    
    def test_coroutine(self):
        """Test avec asyncio.run"""
        async def coro():
            await asyncio.sleep(0.1)
            return 42
        
        result = asyncio.run(coro())
        self.assertEqual(result, 42)

# PYTEST avec pytest-asyncio
# pip install pytest-asyncio
import pytest

@pytest.mark.asyncio
async def test_async_function():
    """Test async avec pytest"""
    result = await my_async_function()
    assert result == expected

# MOCK pour async
from unittest.mock import AsyncMock, patch

async def test_with_mock():
    """Mock de fonctions asynchrones"""
    mock_fetch = AsyncMock(return_value={'data': 'test'})
    
    with patch('module.fetch_data', mock_fetch):
        result = await function_that_calls_fetch()
        assert result == {'data': 'test'}
        mock_fetch.assert_called_once()

# FAKE SLEEP pour tests rapides
async def test_with_fake_sleep():
    """Teste sans délai réel"""
    async def fast_sleep(duration):
        """Sleep instantané pour tests"""
        pass
    
    with patch('asyncio.sleep', fast_sleep):
        # Code qui utilise sleep s'exécute instantanément
        await function_with_sleep()



# 20. MIGRATION DEPUIS CODE SYNCHRONE


# FONCTION SYNCHRONE ORIGINALE
def sync_function():
    """Code bloquant original"""
    time.sleep(1)
    result = requests.get('https://example.com')
    with open('file.txt', 'r') as f:
        data = f.read()
    return result.text + data

# VERSION ASYNCHRONE
async def async_function():
    """Version non-bloquante"""
    import aiohttp
    import aiofiles
    
    await asyncio.sleep(1)
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://example.com') as response:
            result = await response.text()
    
    async with aiofiles.open('file.txt', 'r') as f:
        data = await f.read()
    
    return result + data

# WRAPPER pour code legacy
async def wrap_sync_code():
    """Exécute code sync dans executor"""
    def legacy_function():
        # Code synchrone qu'on ne peut pas modifier
        time.sleep(1)
        return "result"
    
    # Exécute sans bloquer event loop
    result = await asyncio.to_thread(legacy_function)
    return result

# MIGRATION PROGRESSIVE
class HybridService:
    """Service supportant sync et async"""
    
    def process_sync(self, data):
        """Méthode synchrone (legacy)"""
        return self._do_work(data)
    
    async def process_async(self, data):
        """Nouvelle méthode asynchrone"""
        return await asyncio.to_thread(self._do_work, data)
    
    def _do_work(self, data):
        """Logique métier (sync)"""
        time.sleep(1)
        return data.upper()



[OK] 21. BONNES PRATIQUES ET PIÈGES


# [X] PIÈGE: Oublier await
async def common_mistake_1():
    """N'oubliez pas await!"""
    # [X] Mauvais: crée coroutine mais ne l'exécute pas
    result = my_coroutine()
    
    # [OK] Correct: exécute la coroutine
    result = await my_coroutine()

# [X] PIÈGE: Boucle bloquante
async def common_mistake_2():
    """N'utilisez pas time.sleep dans async!"""
    # [X] Mauvais: bloque tout l'event loop
    time.sleep(1)
    
    # [OK] Correct: sleep non-bloquant
    await asyncio.sleep(1)

# [X] PIÈGE: I/O synchrone
async def common_mistake_3():
    """Évitez I/O bloquant"""
    # [X] Mauvais: bloque event loop
    with open('file.txt') as f:
        data = f.read()
    
    # [OK] Correct: I/O asynchrone
    import aiofiles
    async with aiofiles.open('file.txt') as f:
        data = await f.read()

# [X] PIÈGE: Shared state sans lock
shared_counter = 0

async def common_mistake_4():
    """Race condition sans protection"""
    global shared_counter
    
    # [X] Mauvais: race condition
    temp = shared_counter
    await asyncio.sleep(0)  # Autre coroutine peut s'exécuter!
    shared_counter = temp + 1
    
    # [OK] Correct: avec lock
    lock = asyncio.Lock()
    async with lock:
        shared_counter += 1

# [OK] BONNE PRATIQUE: Cleanup avec try/finally
async def good_practice_1():
    """Toujours nettoyer ressources"""
    resource = await acquire_resource()
    try:
        await use_resource(resource)
    finally:
        await release_resource(resource)

# [OK] BONNE PRATIQUE: Context managers
async def good_practice_2():
    """Préférer context managers"""
    async with AsyncResource() as resource:
        await use_resource(resource)
    # Cleanup automatique

# [OK] BONNE PRATIQUE: Timeouts
async def good_practice_3():
    """Toujours timeout les I/O réseau"""
    try:
        result = await asyncio.wait_for(
            network_call(),
            timeout=10.0
        )
    except asyncio.TimeoutError:
        # Gérer timeout
        pass

# [OK] BONNE PRATIQUE: Gestion d'erreurs
async def good_practice_4():
    """Gérer exceptions dans tasks"""
    async def task():
        try:
            await risky_operation()
        except Exception as e:
            logger.error(f"Task failed: {e}")
            raise
    
    tasks = [asyncio.create_task(task()) for _ in range(10)]
    
    # gather avec return_exceptions pour continuer même si échecs
    results = await asyncio.gather(*tasks, return_exceptions=True)

# [OK] BONNE PRATIQUE: Cancellation
async def good_practice_5():
    """Gérer cancellation proprement"""
    try:
        await long_running_operation()
    except asyncio.CancelledError:
        # Cleanup avant propagation
        await cleanup()
        raise  # Toujours re-raise CancelledError



[OK] 22. EXEMPLE COMPLET: WEB SCRAPER


async def complete_example():
    """Scraper web complet avec asyncio"""
    import aiohttp
    from bs4 import BeautifulSoup
    
    # Configuration
    max_concurrent = 5
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []
    
    async def fetch_page(session, url):
        """Fetch page avec rate limiting"""
        async with semaphore:
            try:
                async with session.get(url, timeout=10) as response:
                    html = await response.text()
                    return url, html
            except Exception as e:
                print(f"Error fetching {url}: {e}")
                return url, None
    
    async def parse_page(url, html):
        """Parse HTML (CPU-bound, utiliser executor)"""
        if html is None:
            return None
        
        def parse():
            soup = BeautifulSoup(html, 'html.parser')
            return {
                'url': url,
                'title': soup.title.string if soup.title else None,
                'links': [a['href'] for a in soup.find_all('a', href=True)]
            }
        
        return await asyncio.to_thread(parse)
    
    async def scrape_url(session, url):
        """Scrape une URL complète"""
        url, html = await fetch_page(session, url)
        result = await parse_page(url, html)
        return result
    
    # Liste URLs à scraper
    urls = [
        'https://example.com',
        'https://example.org',
        'https://example.net',
        # ... plus d'URLs
    ]
    
    # Exécution
    async with aiohttp.ClientSession() as session:
        tasks = [scrape_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Traitement résultats
    for result in results:
        if isinstance(result, Exception):
            print(f"Failed: {result}")
        elif result:
            print(f"Scraped: {result['title']}")
    
    return results



[OK] 23. RESSOURCES ET DOCUMENTATION


"""
DOCUMENTATION OFFICIELLE:
- https://docs.python.org/3/library/asyncio.html

BIBLIOTHÈQUES POPULAIRES:
- aiohttp: Client/serveur HTTP async
- aiofiles: I/O fichiers async
- aiomysql / aiopg: Bases de données async
- aioredis: Redis async
- asyncpg: PostgreSQL haute performance
- httpx: Client HTTP moderne (sync + async)

OUTILS:
- pytest-asyncio: Tests async avec pytest
- aiodebug: Outils de debugging
- aiomonitor: Monitoring asyncio

CONCEPTS CLÉS:
1. Coroutines: Fonctions avec async/await
2. Tasks: Enveloppes pour exécution concurrente
3. Event Loop: Ordonnanceur central
4. Futures: Résultats futurs
5. Synchronization: Lock, Event, Semaphore, etc.
6. Queues: Communication entre coroutines
7. Streams: I/O réseau asynchrone

RÈGLES D'OR:
- Toujours await les coroutines
- Jamais de time.sleep() (utiliser asyncio.sleep)
- Éviter I/O bloquant (utiliser versions async)
- Protéger shared state avec locks
- Toujours timeout les I/O réseau
- Gérer CancelledError proprement
- Nettoyer ressources dans finally
"""

# Point d'entrée principal
if __name__ == '__main__':
    # Exécuter application asyncio
    asyncio.run(main())