Boostez les performances de votre API avec la programmation asynchrone dans FastAPI-Tutoriel Python-php.cn

Prêt à faire passer votre jeu API au niveau supérieur ?

FastAPI est là pour rendre vos API plus rapides, plus réactives et capables de gérer de lourdes charges comme un pro.

Dans cet article, nous allons vous montrer comment tirer parti de la programmation asynchrone dans FastAPI pour créer des API hautes performances. À la fin, vous disposerez des connaissances nécessaires pour implémenter des points de terminaison asynchrones et les tester efficacement.

Ce que vous apprendrez

Voici ce que vous maîtriserez :

Les bases de la programmation asynchrone et pourquoi c'est important.
Comment configurer un environnement FastAPI pour le développement asynchrone.
Écrire et tester des points de terminaison asynchrones avec des exemples concrets.
Utilisation de bibliothèques asynchrones pour les requêtes HTTP, la gestion des fichiers et les tâches en arrière-plan.

Pourquoi utiliser la programmation asynchrone dans FastAPI ?

Qu'est-ce que c'est?

La programmation asynchrone permet aux tâches de s'exécuter simultanément. Ceci est particulièrement utile pour les tâches telles que les requêtes réseau, les requêtes de base de données ou les opérations sur les fichiers pour lesquelles l'attente d'une réponse est courante.

Pourquoi est-ce important ?

Dans la programmation synchrone traditionnelle, les tâches s'exécutent de manière séquentielle, ce qui entraîne des retards lors du traitement de plusieurs requêtes. Avec la programmation asynchrone, vous pouvez servir plusieurs utilisateurs simultanément, maximisant l'utilisation des ressources et garantissant une meilleure expérience utilisateur.

Configuration de votre environnement FastAPI

Maintenant, retroussons nos manches et construisons quelque chose d'incroyable !

Tout d'abord, installez les bibliothèques requises :

pip install "fastapi[standard]" httpx aiofiles pytest

Copier après la connexion

Le code

Vous trouverez ci-dessous un exemple complet illustrant la programmation asynchrone dans FastAPI. Chaque partie du code répond à un objectif unique et nous les expliquerons étape par étape.

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

Copier après la connexion

Le décomposer

Points de terminaison de l'API

1.Point de terminaison asynchrone simple

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Copier après la connexion

Ce point de terminaison montre comment définir une route asynchrone de base à l'aide de async def. Il récupère un élément par son identifiant.

2.Appel d'une API externe

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

Copier après la connexion

Cela montre comment effectuer une requête HTTP non bloquante à l'aide de httpx. En attendant la réponse de l'API externe, votre application peut traiter d'autres requêtes.

3.Lecture de fichiers asynchrones

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

Copier après la connexion

Ceci lit un fichier de manière asynchrone, garantissant que l'opération sur le fichier ne bloque pas l'application.

4.Exécution des tâches en arrière-plan

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

Copier après la connexion

Ce point de terminaison planifie une tâche en arrière-plan pour envoyer un e-mail, permettant au fil de discussion principal de gérer d'autres demandes.

Tester le code

1.Test du point de terminaison de base

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

Copier après la connexion

Cela vérifie que le point de terminaison /item/{item_id} renvoie les données attendues.

2.Test de la lecture des fichiers

pip install "fastapi[standard]" httpx aiofiles pytest

Copier après la connexion

Cela crée un fichier de test et vérifie si le point de terminaison /read-file lit et renvoie correctement son contenu.

3.Test de la tâche en arrière-plan

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

Copier après la connexion

Cela teste si la tâche de courrier électronique en arrière-plan est planifiée avec succès.

4.Test de l'appel API externe

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

Copier après la connexion

Cela garantit que le point de terminaison /external-api récupère correctement les données d'une source externe.

Sortir

Supercharge Your API Performance with Asynchronous Programming in FastAPI

Conclusion

Avec le code fourni, vous comprenez désormais de manière pratique comment créer et tester des API asynchrones à l'aide de FastAPI. Qu'il s'agisse de gérer des fichiers, d'appeler des API externes ou de planifier des tâches en arrière-plan, la programmation asynchrone vous permet de créer des applications hautes performances qui évoluent sans effort.

Prêt à créer votre prochain projet FastAPI ? Commençons !

Merci d'avoir lu...
Bon codage !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!