Steigern Sie Ihre API-Leistung mit asynchroner Programmierung in FastAPI-Python-Tutorial-php.cn

Bereit, Ihr API-Spiel auf die nächste Stufe zu bringen?

FastAPI ist hier, um Ihre APIs schneller und reaktionsfähiger zu machen und sie in die Lage zu versetzen, schwere Lasten wie ein Profi zu bewältigen.

In diesem Artikel zeigen wir Ihnen, wie Sie die asynchrone Programmierung in FastAPI nutzen, um leistungsstarke APIs zu erstellen. Am Ende verfügen Sie über das Wissen, asynchrone Endpunkte zu implementieren und effektiv zu testen.

Was Sie lernen werden

Das werden Sie beherrschen:

Die Grundlagen der asynchronen Programmierung und warum sie wichtig ist.
So richten Sie eine FastAPI-Umgebung für die asynchrone Entwicklung ein.
Schreiben und Testen asynchroner Endpunkte anhand von Beispielen aus der Praxis.
Verwendung asynchroner Bibliotheken für HTTP-Anfragen, Dateiverarbeitung und Hintergrundaufgaben.

Warum asynchrone Programmierung in FastAPI verwenden?

Was ist das?

Asynchrone Programmierung ermöglicht die gleichzeitige Ausführung von Aufgaben. Dies ist besonders nützlich für Aufgaben wie Netzwerkanfragen, Datenbankabfragen oder Dateivorgänge, bei denen häufig auf eine Antwort gewartet wird.

Warum ist es wichtig?

Bei der herkömmlichen synchronen Programmierung werden Aufgaben nacheinander ausgeführt, was zu Verzögerungen bei der Bearbeitung mehrerer Anfragen führt. Mit der asynchronen Programmierung können Sie mehrere Benutzer gleichzeitig bedienen, wodurch die Ressourcennutzung maximiert und ein besseres Benutzererlebnis gewährleistet wird.

Einrichten Ihrer FastAPI-Umgebung

Jetzt krempeln wir die Ärmel hoch und bauen etwas Erstaunliches!

Installieren Sie zunächst die erforderlichen Bibliotheken:

pip install "fastapi[standard]" httpx aiofiles pytest

Nach dem Login kopieren

Der Kodex

Unten finden Sie ein vollständiges Beispiel, das die asynchrone Programmierung in FastAPI demonstriert. Jeder Teil des Codes dient einem einzigartigen Zweck und wir erklären ihn Schritt für Schritt.

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()

Nach dem Login kopieren

Aufschlüsseln

API-Endpunkte

1.Einfacher asynchroner Endpunkt

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

Nach dem Login kopieren

Dieser Endpunkt zeigt, wie man mit async def eine grundlegende asynchrone Route definiert. Es ruft ein Element anhand seiner ID ab.

2.Aufrufen einer externen API

@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()

Nach dem Login kopieren

Dies zeigt, wie Sie mit httpx eine nicht blockierende HTTP-Anfrage stellen. Während Sie auf die externe API-Antwort warten, kann Ihre Anwendung andere Anfragen verarbeiten.

3.Asynchrones Lesen von Dateien

@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}

Nach dem Login kopieren

Dadurch wird eine Datei asynchron gelesen, um sicherzustellen, dass der Dateivorgang die Anwendung nicht blockiert.

4.Ausführung von Hintergrundaufgaben

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"}

Nach dem Login kopieren

Dieser Endpunkt plant eine Hintergrundaufgabe zum Senden einer E-Mail, sodass der Hauptthread andere Anfragen bearbeiten kann.

Testen des Codes

1.Basisendpunkt testen

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

Nach dem Login kopieren

Dadurch wird überprüft, ob der Endpunkt /item/{item_id} die erwarteten Daten zurückgibt.

2.Testen des Dateilesens

pip install "fastapi[standard]" httpx aiofiles pytest

Nach dem Login kopieren

Dadurch wird eine Testdatei erstellt und überprüft, ob der /read-file-Endpunkt seinen Inhalt korrekt liest und zurückgibt.

3.Hintergrundaufgabe testen

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()

Nach dem Login kopieren

Dies testet, ob die Hintergrund-E-Mail-Aufgabe erfolgreich geplant wurde.

4.Externen API-Aufruf testen

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

Nach dem Login kopieren

Dadurch wird sichergestellt, dass der /external-api-Endpunkt Daten korrekt von einer externen Quelle abruft.

Ausgabe

Supercharge Your API Performance with Asynchronous Programming in FastAPI

Abschluss

Mit dem bereitgestellten Code verfügen Sie nun über ein praktisches Verständnis dafür, wie Sie asynchrone APIs mit FastAPI erstellen und testen. Ob es darum geht, Dateien zu verarbeiten, externe APIs aufzurufen oder Hintergrundaufgaben zu planen – mit der asynchronen Programmierung können Sie leistungsstarke Anwendungen erstellen, die sich mühelos skalieren lassen.

Bereit, Ihr nächstes FastAPI-Projekt zu erstellen? Fangen wir an!

Danke fürs Lesen...
Viel Spaß beim Codieren!

Das obige ist der detaillierte Inhalt vonSteigern Sie Ihre API-Leistung mit asynchroner Programmierung in FastAPI. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!