Tingkatkan Prestasi API Anda dengan Pengaturcaraan Asynchronous dalam FastAPI

Bersedia untuk membawa permainan API anda ke peringkat seterusnya?

FastAPI sedia untuk menjadikan API anda lebih pantas, lebih responsif dan mampu mengendalikan beban berat seperti seorang profesional.

Dalam artikel ini, kami akan menunjukkan kepada anda cara memanfaatkan pengaturcaraan tak segerak dalam FastAPI untuk membina API berprestasi tinggi. Pada akhirnya, anda akan dilengkapi dengan pengetahuan untuk melaksanakan titik akhir tak segerak dan mengujinya dengan berkesan.

Apa yang Anda Akan Pelajari

Ini perkara yang anda akan kuasai:

• Asas pengaturcaraan tak segerak dan sebab ia penting.
• Cara menyediakan persekitaran FastAPI untuk pembangunan async.
• Menulis dan menguji titik akhir tak segerak dengan contoh dunia sebenar.
• Menggunakan perpustakaan async untuk permintaan HTTP, pengendalian fail dan tugasan latar belakang.

Mengapa Menggunakan Pengaturcaraan Asynchronous dalam FastAPI?

Apa Itu?

Pengaturcaraan tak segerak membolehkan tugasan dijalankan serentak. Ini amat berguna untuk tugas seperti permintaan rangkaian, pertanyaan pangkalan data atau operasi fail yang menunggu respons adalah perkara biasa.

Mengapa Ia Penting?

Dalam pengaturcaraan segerak tradisional, tugasan dijalankan secara berurutan, membawa kepada kelewatan apabila mengendalikan berbilang permintaan. Dengan pengaturcaraan tak segerak, anda boleh melayani berbilang pengguna secara serentak, memaksimumkan penggunaan sumber dan memastikan pengalaman pengguna yang lebih baik.

Menyediakan Persekitaran FastAPI Anda

Sekarang, mari kita menyingsing lengan baju kita dan membina sesuatu yang menakjubkan!

Mula-mula, pasang perpustakaan yang diperlukan:

pip install "fastapi[standard]" httpx aiofiles pytest

Kod

Di bawah ialah contoh lengkap yang menunjukkan pengaturcaraan tak segerak dalam FastAPI. Setiap bahagian kod mempunyai tujuan yang unik, dan kami akan menerangkannya langkah demi langkah.

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

Memecahkannya

Titik Akhir API

1.Titik Akhir Async Mudah

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

Titik akhir ini mempamerkan cara untuk menentukan laluan tak segerak asas menggunakan async def. Ia mendapatkan semula item dengan IDnya.

2.Memanggil API Luaran

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

Ini menunjukkan cara membuat permintaan HTTP tidak menyekat menggunakan httpx. Semasa menunggu respons API luaran, aplikasi anda boleh memproses permintaan lain.

3.Bacaan Fail Tak Segerak

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

Ini membaca fail secara tidak segerak, memastikan operasi fail tidak menyekat aplikasi.

4.Pelaksanaan Tugas Latar Belakang

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

Titik akhir ini menjadualkan tugas latar belakang untuk menghantar e-mel, membenarkan urutan utama mengendalikan permintaan lain.

Menguji Kod

1.Menguji Titik Akhir Asas

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

Ini mengesahkan titik akhir /item/{item_id} mengembalikan data yang dijangkakan.

2.Menguji Pembacaan Fail

pip install "fastapi[standard]" httpx aiofiles pytest

Ini mencipta fail ujian dan menyemak sama ada titik akhir /read-file membaca dan mengembalikan kandungannya dengan betul.

3.Tugas Latar Belakang Menguji

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

Ini menguji sama ada tugasan e-mel latar belakang berjaya dijadualkan.

4.Menguji Panggilan API Luaran

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

Ini memastikan titik akhir /external-api mengambil data daripada sumber luaran dengan betul.

Keluaran

Kesimpulan

Dengan kod yang disediakan, anda kini mempunyai pemahaman praktikal tentang cara membina dan menguji API tak segerak menggunakan FastAPI. Sama ada mengendalikan fail, memanggil API luaran atau menjadualkan tugas latar belakang, pengaturcaraan tak segerak membolehkan anda mencipta aplikasi berprestasi tinggi yang berskala dengan mudah.

Bersedia untuk membina projek FastAPI anda yang seterusnya? Mari mulakan!

Terima kasih kerana membaca...
Selamat Pengekodan!

Atas ialah kandungan terperinci Tingkatkan Prestasi API Anda dengan Pengaturcaraan Asynchronous dalam FastAPI. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!