0Pricing
Python Academy · Leçon

Schémas d’entrées-sorties asynchrones en production

Appliquez asyncio aux clients HTTP, aux files d’attente et aux pipelines utilisés en conditions réelles.

Schémas d’entrées-sorties asynchrones en production est une leçon Python Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage Python Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours Python Academy comprend 4 leçons au total.

Bonnes pratiques pour la boucle d’événements

Utilisez toujours asyncio.run() comme point d’entrée unique. Évitez de mélanger négligemment le code synchrone et asynchrone, et ne bloquez jamais la boucle d’événements avec un long calcul effectué par le processeur.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Mutualisation des connexions

Réutilisez les connexions au lieu d’en ouvrir une nouvelle pour chaque requête. Des bibliothèques comme asyncpg et aiohttp fournissent des pools intégrés.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Producteur-consommateur avec asyncio.Queue

asyncio.Queue désolidarise les producteurs et les consommateurs qui s’exécutent sous forme de tâches concurrentes.

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

Limitation du débit avec Semaphore

Limitez le nombre de requêtes sortantes simultanées afin d’éviter de surcharger les API externes ou d’atteindre leurs limites de débit.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

Nouvelles tentatives avec temporisation exponentielle

Entourez les opérations pouvant être réessayées d’une logique de temporisation afin de gérer correctement les défaillances temporaires.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Arrêt progressif

Interceptez asyncio.CancelledError dans les tâches de longue durée et gérez les signaux du système d’exploitation pour effectuer un arrêt propre.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Réponses en flux

Utilisez des générateurs asynchrones pour transmettre des réponses volumineuses sans les charger entièrement en mémoire.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Tâches d’arrière-plan dans FastAPI

Dans FastAPI, BackgroundTasks met en file d’attente le travail à exécuter après l’envoi de la réponse, ce qui permet de conserver des points d’accès rapides.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Modèle de signal de présence

Exécutez une tâche d’arrière-plan périodique parallèlement à la boucle principale du serveur avec create_task.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Profilage du code asynchrone

Utilisez yappi ou py-spy pour profiler du code asynchrone. Activez le mode de débogage d’asyncio pour recevoir des avertissements concernant les rappels lents.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Éviter les pièges courants

Règles clés : n’appelez jamais de code bloquant dans la boucle d’événements ; attendez toujours les coroutines ; annulez les tâches à l’arrêt ; utilisez des pools de connexions ; limitez la concurrence avec des sémaphores.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Vérification rapide

Quelle primitive d’asyncio désolidarise les producteurs et les consommateurs qui s’exécutent sous forme de tâches concurrentes ?

Récapitulatif

Modèles asynchrones pour la production : utilisez des pools de connexions, contrôlez la concurrence avec Semaphore, implémentez les nouvelles tentatives et la temporisation, diffusez les données volumineuses en flux, effectuez un arrêt progressif en réponse aux signaux et exécutez les tâches périodiques avec create_task.

Questions Fréquemment Posées

La leçon « Schémas d’entrées-sorties asynchrones en production » est-elle gratuite ?

Oui — le texte complet de « Schémas d’entrées-sorties asynchrones en production » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours Python Academy, passe à CoddyKit PRO. Le cours Python Academy comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Schémas d’entrées-sorties asynchrones en production » ?

Appliquez asyncio aux clients HTTP, aux files d’attente et aux pipelines utilisés en conditions réelles. Tu pratiques Python Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer Python Academy ?

Aucune expérience préalable n'est requise. Python Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.

Combien de temps prend la leçon « Schémas d’entrées-sorties asynchrones en production » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon Python Academy ?

Oui. Chaque leçon Python Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Coroutines et boucle d’événements
  2. await, tâches et rassemblement
  3. Gestionnaires de contexte et itérateurs asynchrones
  4. Schémas d’entrées-sorties asynchrones en production
← Retour à Python Academy