Menguasai Python Async IO dengan FastAPI

Memandangkan Python ialah bahasa yang ditafsirkan, apabila digunakan untuk pembangunan back-end, seperti dalam gabungan Python Django, berbanding Java Spring, masa tindak balasnya akan menjadi lebih lama sedikit. Walau bagaimanapun, selagi kod itu munasabah, perbezaannya tidak terlalu ketara. Walaupun Django menggunakan mod berbilang proses, keupayaan pemprosesan serentaknya masih lebih lemah. Python mempunyai beberapa penyelesaian untuk meningkatkan keupayaan pemprosesan serentak. Contohnya, menggunakan rangka kerja tak segerak FastAPI, dengan keupayaan tak segeraknya, keupayaan pemprosesan serentak tugas intensif I/O boleh dipertingkatkan dengan banyak. FastAPI ialah salah satu rangka kerja Python terpantas.

FastAPI sebagai contoh

Mari kita lihat dahulu secara ringkas cara menggunakan FastAPI.

Contoh 1: Rangkaian Lalai Asynchronous IO

Pemasangan:

pip install fastapi

Kod Bahagian Pelayan Mudah:

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}

Permulaan:

uvicorn app:app --reload

Kita dapat melihat bahawa, berbanding dengan rangka kerja lain, antara muka FastAPI hanya mempunyai kata kunci async tambahan. Kata kunci async mentakrifkan antara muka sebagai tak segerak. Daripada hasil pulangan sahaja, kami tidak dapat membezakan antara FastAPI dan rangka kerja Python yang lain. Perbezaannya terletak pada akses serentak. Apabila rangkaian pelayan FastAPI mengendalikan permintaan laluan, seperti http://127.0.0.1:8000/, jika mereka menemui rangkaian I/O, mereka tidak akan menunggu lagi tetapi mengendalikan permintaan lain sebaliknya. Apabila I/O rangkaian selesai, pelaksanaan akan disambung semula. Keupayaan tak segerak ini meningkatkan keupayaan pemprosesan tugas intensif I/O.

Contoh 2: IO Asynchronous Rangkaian Eksplisit

Mari kita lihat contoh lain. Dalam kod perniagaan, permintaan rangkaian tak segerak yang jelas dimulakan. Untuk I/O rangkaian ini, sama seperti permintaan laluan, FastAPI juga akan mengendalikannya secara tak segerak.

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()

Jika anda mahu pangkalan data I/O menjadi tak segerak, anda memerlukan sokongan operasi tak segerak daripada pemacu pangkalan data atau ORM.

IO tak segerak

Pelaksanaan teras tak segerak FastAPI ialah I/O tak segerak. Kami boleh memulakan pelayan dengan keupayaan pemprosesan tak segerak secara terus menggunakan I/O tak segerak tanpa menggunakan FastAPI.

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')

async def init(loop):
    # Use the event loop to monitor web requests
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    # Start the server, and the event loop monitors and processes web requests
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

# Explicitly get an event loop
loop = asyncio.get_event_loop()
# Start the event loop
loop.run_until_complete(init(loop))
loop.run_forever()

Apabila contoh ini dimulakan, hasil pulangan http://127.0.0.1:8000/ adalah sama seperti Contoh 1. Prinsip pelaksanaan asas I/O tak segerak ialah "coroutines" dan "event loops" .

Coroutines

pip install fastapi

Indeks fungsi ditakrifkan dengan async def, yang bermaksud ia adalah coroutine. Kata kunci await digunakan sebelum operasi I/O untuk memberitahu utas pelaksanaan supaya tidak menunggu operasi I/O ini. Panggilan fungsi biasa dilaksanakan melalui timbunan, dan fungsi hanya boleh dipanggil dan dilaksanakan satu demi satu. Walau bagaimanapun, coroutine ialah sejenis fungsi khas (bukan benang kolaboratif). Ia membenarkan utas untuk menjeda pelaksanaan pada tanda menunggu dan beralih untuk melaksanakan tugas lain. Apabila operasi I/O selesai, pelaksanaan akan diteruskan.

Mari kita lihat kesan berbilang coroutine yang dilaksanakan serentak.

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}

Output:

uvicorn app:app --reload

Kita dapat melihat bahawa utas tidak melaksanakan tiga tugasan satu demi satu. Apabila ia menghadapi operasi I/O, ia bertukar untuk melaksanakan tugas lain. Selepas operasi I/O selesai, ia terus dilaksanakan. Ia juga boleh dilihat bahawa tiga coroutine pada asasnya mula menunggu operasi I/O pada masa yang sama, jadi masa penyiapan pelaksanaan akhir pada dasarnya adalah sama. Walaupun gelung peristiwa tidak digunakan secara eksplisit di sini, asyncio.run akan menggunakannya secara tersirat.

Penjana

Coroutine dilaksanakan melalui penjana. Penjana boleh menjeda pelaksanaan fungsi dan juga menyambung semula, yang merupakan ciri coroutine.

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()

Apabila menjalankan penjana dengan next(), apabila ia menemui hasil, ia akan berhenti seketika. Apabila next() dijalankan semula, ia akan terus berjalan dari hasil di mana ia dijeda kali terakhir. Sebelum Python 3.5, coroutine juga ditulis dengan hasil "anotasi". Bermula daripada Python 3.5, async def await digunakan.

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')

async def init(loop):
    # Use the event loop to monitor web requests
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    # Start the server, and the event loop monitors and processes web requests
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

# Explicitly get an event loop
loop = asyncio.get_event_loop()
# Start the event loop
loop.run_until_complete(init(loop))
loop.run_forever()

Ciri jeda dan sambung semula penjana boleh digunakan untuk banyak perkara selain coroutine. Sebagai contoh, ia boleh mengira semasa menggelung dan menyimpan algoritma. Contohnya, melaksanakan segi tiga Pascal (kedua-dua hujung setiap baris ialah 1 dan nombor dalam kedudukan lain ialah hasil tambah dua nombor di atasnya).

async def index(request):
    await asyncio.sleep(1)  # Simulate I/O operation
    return web.Response(text='{"Hello": "World"}', content_type='application/json')

Output:

import asyncio
from datetime import datetime

async def coroutine3():
    print(f"Coroutine 3 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 3 finished at {datetime.now()}")

async def coroutine2():
    print(f"Coroutine 2 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 2 finished at {datetime.now()}")

async def coroutine1():
    print(f"Coroutine 1 started at {datetime.now()}")
    await asyncio.sleep(1)  # Simulate I/O operation
    print(f"Coroutine 1 finished at {datetime.now()}")

async def main():
    print("Main started")

    # Create tasks to make coroutines execute concurrently
    task1 = asyncio.create_task(coroutine1())
    task2 = asyncio.create_task(coroutine2())
    task3 = asyncio.create_task(coroutine3())

    # Wait for all tasks to complete
    await task1
    await task2
    await task3

    print("Main finished")

# Run the main coroutine
asyncio.run(main())

Gelung Acara

Memandangkan pelaksanaan coroutine boleh dijeda, bilakah coroutine akan menyambung semula pelaksanaan? Ini memerlukan penggunaan gelung peristiwa untuk memberitahu urutan pelaksanaan.

Main started
Coroutine 1 started at 2024-12-27 12:28:01.661251
Coroutine 2 started at 2024-12-27 12:28:01.661276
Coroutine 3 started at 2024-12-27 12:28:01.665012
Coroutine 1 finished at 2024-12-27 12:28:02.665125
Coroutine 2 finished at 2024-12-27 12:28:02.665120
Coroutine 3 finished at 2024-12-27 12:28:02.665120
Main finished

Gelung acara menggunakan teknologi pemultipleksan I/O, sentiasa berbasikal untuk memantau acara di mana coroutine boleh terus dilaksanakan. Apabila ia boleh dilaksanakan, utas akan terus melaksanakan coroutine.

Teknologi Multiplexing I/O

Untuk memahami pemultipleksan I/O dengan cara yang mudah: Saya bos stesen kurier. Saya tidak perlu bertanya secara aktif kepada setiap kurier tentang penyelesaian tugas mereka. Sebaliknya, kurier akan datang kepada saya sendiri selepas menyelesaikan tugas mereka. Ini meningkatkan keupayaan pemprosesan tugas saya dan saya boleh melakukan lebih banyak perkara.

pilih, tinjauan pendapat dan epoll semuanya boleh mencapai pemultipleksan I/O. Berbanding dengan pilihan dan tinjauan pendapat, epoll mempunyai prestasi yang lebih baik. Linux biasanya menggunakan epoll secara lalai, dan macOS menggunakan kqueue, yang serupa dengan epoll dan mempunyai prestasi yang serupa.

Pelayan Soket Menggunakan Gelung Acara

pip install fastapi

Mulakan soket pelayan untuk memantau port yang ditentukan. Jika berjalan pada sistem Linux, pemilih menggunakan epoll sebagai pelaksanaannya secara lalai. Kod menggunakan epoll untuk mendaftarkan acara penerimaan permintaan (acara terima). Apabila permintaan baharu tiba, epoll akan mencetus dan melaksanakan fungsi pengendalian acara, dan pada masa yang sama, mendaftarkan acara baca (peristiwa baca) untuk memproses dan membalas data permintaan. Apabila diakses dari bahagian web dengan http://127.0.0.1:8000/, hasil pemulangan adalah sama seperti Contoh 1. Log berjalan pelayan:

# app.py
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}

Pelayan Soket

Gunakan Soket secara langsung untuk memulakan pelayan. Apabila diakses dengan penyemak imbas di http://127.0.0.1:8080/ atau menggunakan curl http://127.0.0.1:8080/, ia akan mengembalikan {"Hello": "World"}

uvicorn app:app --reload

Apabila diakses dengan curl http://127.0.0.1:8001/, Log berjalan pelayan:

# app.py
from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

# Example of an asynchronous GET request
@app.get("/external-api")
async def call_external_api():
    url = "https://leapcell.io"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code!= 200:
            raise HTTPException(status_code=response.status_code, detail="Failed to fetch data")
        return response.json()

Ringkasan

I/O tak segerak dilaksanakan di lapisan bawah menggunakan "coroutines" dan "gelung peristiwa". "Coroutines" memastikan bahawa apabila utas menemui operasi I/O bertanda semasa pelaksanaan, ia tidak perlu menunggu I/O selesai tetapi boleh menjeda dan membiarkan utas melaksanakan tugas lain tanpa menyekat. "Gelung peristiwa" menggunakan teknologi pemultipleksan I/O, sentiasa berbasikal untuk memantau acara I/O. Apabila acara I/O tertentu selesai, panggilan balik yang sepadan dicetuskan, membolehkan coroutine meneruskan pelaksanaan.

---

Leapcell: Platform Ideal untuk FastAPI dan Aplikasi Python Lain:

Akhir sekali, izinkan saya memperkenalkan platform yang ideal untuk menggunakan Flask/FastAPI: Leapcell.

Leapcell ialah platform pengkomputeran awan yang direka khusus untuk aplikasi pengedaran moden. Model penetapan harga bayar semasa anda memastikan tiada kos terbiar, bermakna pengguna hanya membayar untuk sumber yang sebenarnya mereka gunakan.

Kelebihan unik Leapcell untuk aplikasi WSGI/ASGI:

1. Sokongan Pelbagai Bahasa

• Menyokong pembangunan dalam JavaScript, Python, Go atau Rust.

2. Penggunaan Percuma Projek Tanpa Had

• Hanya caj berdasarkan penggunaan. Tiada caj apabila tiada permintaan.

3. Keberkesanan Kos yang tiada tandingan

• Bayar semasa anda pergi, tanpa yuran terbiar.
• Sebagai contoh, $25 boleh menyokong 6.94 juta permintaan, dengan purata masa tindak balas 60 milisaat.

4. Pengalaman Pembangun Dipermudahkan

• Antara muka pengguna yang intuitif untuk persediaan mudah.
• Saluran paip CI/CD automatik sepenuhnya dan penyepaduan GitOps.
• Metrik dan log masa nyata, memberikan cerapan yang boleh diambil tindakan.

5. Kebolehskalaan dan Prestasi Tinggi yang Mudah

• Penskalaan automatik untuk mengendalikan konkurensi tinggi dengan mudah.
• Sifar operasi overhed, membolehkan pembangun menumpukan pada pembangunan.

Ketahui lebih lanjut dalam dokumentasi!

Twitter Leapcell: https://x.com/LeapcellHQ

Atas ialah kandungan terperinci Menguasai Python Async IO dengan FastAPI. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!