Pertanyaan dan solusi yang sering diajukan saat membangun server fastapi – Beragampengetahuan

Ketika proyek dimulai, fungsionalitas lebih diutamakan daripada struktur. Ketika kompleksitas tumbuh, menjadi semakin sulit untuk menemukan kode titik akhir, dan impor berulang dan loop secara bertahap menyebar. Sementara struktur yang tepat bervariasi dengan kebutuhan bisnis, sebagian besar server web berbagi kekhawatiran yang sama dan dapat terdiri dari folder berikut dan dipisahkan sesuai dengan fungsionalitas:

Main.py – Buat aplikasi; kabel seumur hidup; Daftarkan Perangkat Lunak Menengah dan Penangan Kesalahan.

Verifier/ – Pemeriksaan input yang dapat digunakan kembali (ID, enumerasi, aturan bisnis) yang digunakan oleh Pydantic.
middleware/ – Lapisan silang (mis., Penjaga ukuran muatan).

error_handlers/ – Pengecualian Pemetaan Respons HTTP;
Utilitas/ – Asisten kecil (kertas pembungkus responsif, paging, format).
model/ – Model Pydantic untuk memvalidasi permintaan yang masuk dan tanggapan keluar;

Melayani/ – Logika Bisnis; merencanakan repositori/pelanggan.

klien/ – Klien: Mesin/pembuat sesi DB, klien HTTP, adaptor rekaman/pelacakan.

pabrik/ – Suntikan pembangun yang ramah untuk dependensi pelanggan/layanan.

Titik akhir/ – dikelompokkan berdasarkan domain bisnis (akun/, pembelian/, pesanan/);
tes/ – Uji unit/integrasi; Gunakan override dan pengaturan pengujian.

Bagian yang rumit bukanlah foldernya – itu adalah hubungan router orang tua/anak.

Mari bertujuan untuk desain API ini.

/accounts
/accounts/account_id
/accounts/account_id/orders
/accounts/account_id/orders/order_id

Untuk mencapai ini, kami akan membuat dua router:

Tempat: Aplikasi/titik akhir/akun/pesanan/root.py

Melayani: /CoCOCOR_ID/PESANAN

orders = APIRouter(
    prefix="/account_id/orders", 
    tags=["orders"]
)

@orders.get("
async def list_orders(
    account_id: Annotated[UUID, Path()], 
    last_id: str | None = None, 
    limit: int = 50
):
    return 
      "account": str(account_id), 
      "items": [], 
      "next": None
    

@orders.post("
async def create_order(
    account_id: Annotated[UUID, Path()]
): return  "account": str(account_id), "ok": True

Tempat: Aplikasi/titik akhir/akun/root.py

Melayani: /akun

from .orders.root import orders as orders_router

accounts = APIRouter(
    prefix="/accounts", 
    tags=["accounts"]
)

async def load_account(
    account_id: Annotated[UUID, Path()]
): return "id": account_id

@accounts.get("
async def list_accounts(): return ["id": "12345"]

@accounts.get("/account_id")
async def get_account(account_id: UUID):
    return "id": account_id

accounts.include_router(
    orders_router,
    # DI imposed on the order routes
    dependencies=[Depends(load_account)],
)

Struktur ini memastikan bahwa tidak ada yang ditentukan account_idIni memungkinkan pengembang untuk dengan cepat mengidentifikasi lokasi titik akhir dan membuat router anak-anak berlapis-lapis. Dengan desain ini, “router pesanan” dapat digunakan kembali di tempat lain dengan ketergantungan penyesuaian.

Pabrik merangkum pembangunan pelanggan dan layanan dengan menggunakan status aplikasi dan berbagai variabel dan konfigurasi lingkungan.

Pabrik dapat digunakan untuk mengelola:

1. Klien Cloud – SQS, S3, SNS, dll.
2. Klien Database – Redis, Postgres, Mongo
3. Kabel Konfigurasi – Level Aplikasi dan Tingkat Permintaan (jika setiap pelanggan memiliki konfigurasi tertentu)
4. Kelas Layanan, Logika Bisnis Dienkapsulasi – Layanan Akun, Layanan Pesanan.
5. Kelas Auxiliary – Perekam, Rapat Pengguna, dll.

class Factory:
    @classmethod
    async def get_http_client(
        cls, 
        settings
    ): return AsyncClient(base_url=settings.api)

    @classmethod
    async def get_account_service(
      cls, 
      request: Request
    ) -> AccountService:
      settings: Settings = request.app.state.settings 
      return AccountService(
          client=await cls.get_http_client(settings),
      )

Berdasarkan contoh ini, kita melihat pabrik yang dirancang dengan baik yang dapat dimasukkan ke dalam injeksi ketergantungan Fastapi dan menyisipkan dengan upaya minimal.

@app.get("/accounts/account_id")
async def account_overview(
    account_id: str,
    account_service: Annotated[
        AccountService,
        Depends(Factory.get_account_service)
    ]
):
    r = await account_service.fetch_profile(account_id)
    if not r:
        raise HTTPException(
            status_code=404, 
            detail="Account not found"
        )
    return r

Titik akhir tetap tipis dan dapat dibaca: memvalidasi input, mengoordinasikan layanan dan memetakan hasil internal ke respons HTTP. Logika bisnis hidup di luar titik akhir; Penangan fokus pada masalah HTTP (kabel, kode status, dan pemformatan).

Pabrik dapat sangat menyederhanakan akses ke berbagai login di server.

class Factory
    @classmethod
    def get_root_logger(cls) -> logging.Logger:
        return logging.getLogger("app")

    @classmethod
    def get_request_logger(
        cls, 
        request: Request
     ) -> logging.LoggerAdapter:
        base = cls.get_root_logger()
        return logging.LoggerAdapter(
            base,
            
              "path": request.url.path, 
              "method": request.method,
              "corr_id": request.state.corr_id,
            
        )

Semua jenis perekam sekarang dapat dengan mudah dipakai pada titik akhir melalui pabrik:

acc_ser_dep = Annotated[
    AccountService,
    Depends(Factory.get_account_service),
]
log_dep = Annotated[
    LoggerAdapter,
    Depends(Factory.get_request_logger),
]

@app.get("/accounts/account_id")
async def account_overview(
    account_id: str,
    account_service: acc_ser_dep,
    logger: log_dep,
):
    r = await account_service.fetch_profile(account_id)
    if not r:
        logger.warning(
            "Account not found",
            extra="account_id": account_id,
        )
        raise HTTPException(404, "Account not found")
    return r

Logger yang diminta membutuhkan ID yang relevan untuk menjahit login yang diminta ke dalam cerita yang koheren. Menghasilkan titik akhir internal ID adalah umum, tetapi ini dapat menyebabkan duplikasi. Menggunakan middleware kecil yang mengatur ID sekali per permintaan lebih efisien dan dapat diperluas untuk dimasukkan user_id,,,,, firm_iddan konteks lainnya:

@app.middleware("http")
async def corr_middleware(request: Request, call_next):
    _id = request.headers.get("X-Request-ID") or str(uuid4())
    request.state.corr_id = _id

    response = await call_next(request)
    response.headers["X-Request-ID"] = _id
    return response

Kembali ke contoh pabrik, ada langkah yang mudah dilewatkan untuk menyederhanakan arsitektur injeksi ketergantungan.

settings: Settings = request.app.state.settings

Baris ini mengasumsikan bahwa pengaturan diinisialisasi pada startup (main.py):

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.settings = Settings()
    yield
    await Factory.aclose()


app = FastAPI(lifespan=lifespan)

Objek terlampir app.state Untuk setiap proses kerja FASTAPI, umur harus ditutup dan harus ditutup selama fase shutdown.

Objek seumur hidup dapat bertahan dalam banyak permintaan dan bermain selama masa pakai server. menggunakan app.state Bukan satu -satunya cara untuk menahannya (singleton, objek cache dengan kunci, dll.), Tetapi itu adalah opsi yang paling nyaman di Fastapi.

Ringkasan “sesi” permintaan, yang mengharuskan itu khusus untuk keadaan khusus permintaan (mis., Correlation_id, user_id) dan akan mati setelah permintaan selesai.

Sepanjang kehidupan server, kita dapat menggunakan objek yang terlampir pada app.state (db pool, klien http, dll.). Saat server dimatikan, kami memicu shutdown objek -objek ini dan memberikan logika khusus untuk ini.

class Factory:
    @classmethod
    async def aclose(cls):
        await app.state.db_pool.aclose()
        await app.state.external_http_client.aclose()
        ...

Fakta bahwa suatu objek adalah sampah yang dikumpulkan tidak berarti bahwa koneksi yang sesuai telah ditutup. Itulah mengapa kami memiliki await Factory.aclose() Selama kehidupan server.

Pola yang sama dapat diterapkan pada klien yang meminta melalui pabrik.

class Factory:
    @classmethod
    async def get_account_service(cls):
        settings: Settings = request.app.state.settings

        account_service = AccountService(
            client=await cls.get_http_client(settings),
        )

        try:
            yield account_service
        finally:
            await account_service.client.aclose()

Dalam contoh di atas account_service Hapus hanya setelah menutup koneksi ke klien HTTP sehingga soketnya segera dilepaskan.

Pengaturan adalah contoh yang baik dari objek seumur hidup, dan permintaan dapat diakses secara bersamaan melalui app.state.settings. Idealnya, objek kehidupan harus dibaca hanya dan tidak boleh merujuk pada objek yang dipartisi yang diminta (sehingga sampah dapat dikumpulkan).

Kalau tidak, ada risiko kebocoran memori dan kondisi rasial. Secara umum, apa pun yang dibagikan di seluruh permintaan harus memiliki pelindung bersamaan bawaan (keamanan utas/asinkron).

Untuk berbagi keadaan antara permintaan yang berbeda dan bahkan pekerja, penyimpanan eksternal adalah opsi yang lebih aman dan lebih dapat diskalakan. Tetapi apa yang terjadi jika dua permintaan dari pekerja yang sama mencoba menulis pengaturan di sekitar pengaturan pada saat yang sama? Mari kita pertimbangkan contoh berikut:

cfg = request.app.state.settings
old = cfg.threshold
await some_async_call()  # yields, other requests run here
request.app.state.settings.threshold = old + 1

Kode ini akan membuat keadaan bersama tidak konsisten. Jika pembaruan status benar -benar diperlukan, maka kunci harus digunakan

app.state.settings_lock = asyncio.Lock()

async def update_settings(app, patch: dict):
    async with app.state.settings_lock:
        settings = app.state.settings
        app.state.settings = settings.model_copy(
            update=patch,
        )

Dengan fitur ini, Anda dapat memperbarui pengaturan dengan aman:

await update_settings(request.app,"threshold":old+1)

Mari kita pertimbangkan tidak ada contoh await some_async_call()

cfg = request.app.state.settings
old = cfg.threshold
request.app.state.settings.threshold = old + 1

Kode ini berjalan sebagai satu blok dari utas loop acara. Ini memastikan integritas negara, tetapi meningkatkan latensi server. FastAPI tidak secara otomatis memindahkan pekerjaan sinkron dari utas loop acara ke ThreadPool. Itu harus dilakukan secara eksplisit oleh pengembang.

@app.get("/update-state")
async def update_state():
    # moves to threadpool
    await anyio.to_thread.run_sync(
        update_settings_sync
    )
    return "ok": True

Ada cara lain untuk melakukannya dengan ketergantungan, tetapi ini tidak layak dijelajahi secara rinci, karena memiliki banyak utas menulis ke status bersama bukanlah desain yang tepat.

Kedua contoh pekerjaan sinkronisasi uninstall berfungsi dengan baik ketika tidak ada pembaruan negara yang terlibat.

@app.post("/update-state")
async def update_state(
    result: dict = Depends(update_settings_dep)
):
    return result

Pemahaman yang dapat diandalkan tentang bagaimana loop peristiwa berinteraksi dengan kode sinkron dan asinkron membantu pengembang menemukan solusi yang paling efektif.

Kelelahan sering terjadi di cincin acara asinkron Kinerja titik akhir sangat berat sinkronis Bekerja. Lihat contoh berikut:

def cpu_heavy(n: int) -> float:
    # Python CPU; never yields
    s = 0.0
    for i in range(n):
        s += math.sqrt(i)
    return s

@app.get("/cpu")
async def cpu(n: int = 10_000_000):
    # heavy CPU on the loop, requests on the worker stall
    return "sum": cpu_heavy(n)

@app.get("/sleep")
async def sleep(ms: int = 500):
    # blocking sleep holds the loop
    time.sleep(ms / 1000)
    return "slept_ms": ms

@app.get("/io")
async def io():
    # stalling call until the socket completes
    r = requests.get(
        "
        timeout=5,
    )
    return "status": r.status_code

Menipisnya cincin acara sulit didiagnosis. Pengembang harus melihat permintaan apa yang terjadi ketika permintaan yang terhenti terjadi dan kemudian menelusuri logika untuk mengidentifikasi kemungkinan pelakunya.

Biasanya, fungsi LOOP_LAG khusus dibuat di server (via Lifespan) untuk melaporkan lag yang diperluas ke opentelemetry, tetapi sinyal ini saja jarang menemukan akar penyebabnya.

async def lag_probe(
    interval: float = 0.5,
    warn_ms: float = 100.0
):
    loop = asyncio.get_running_loop()
    next_t = loop.time() + interval

    while True:
        await asyncio.sleep(interval)
        now = loop.time()
        lag_ms = max(0.0, (now - next_t) * 1000.0)

        if lag_ms > warn_ms:
            print(f"event_loop_lag_ms=lag_ms:.1f")

        next_t += interval

Anti-pola khas di Fastapi adalah bahwa satu titik akhir memanggil yang lain. Motivasi yang biasa adalah untuk menghindari menyalin kode yang telah dibagi di tempat lain dan kertas penerbitan tanpa antarmuka layanan yang tepat. Pola ini menambah latensi dan overhead yang tidak perlu – serialisasi tambahan, otentikasi, dan perekaman.

Jika anti-mode diterapkan secara sistematis di seluruh titik akhir, itu akan dengan cepat keluar dari kontrol, memperkuat beban dan penundaan. Karena sifatnya yang eksponensial, menjadi lebih sulit untuk membangun strategi penskalaan berdasarkan jumlah permintaan yang masuk. Tak perlu dikatakan, seiring berjalannya waktu, bahkan kesalahan desain kecil, bola salju terjebak dalam masalah besar.

Anti-mode lainnya adalah memutar banyak pekerja di server, yang bergantung pada struktur memori yang berat (LLM, model Pydantic besar, truk besar, dll.). Setiap pekerja adalah proses terpisah dengan loop acara sendiri, sehingga setiap pekerja menyalin objek -objek ini, meningkatkan memori dan konsumsi CPU. Jika seorang pekerja melebihi batas wadah, pod akan dibunuh (misalnya, Kubernett) sebelum Gungunicorn dapat mendaur ulang pekerja.

Untuk beban kerja ini, inferensi yang tidak muat lebih disukai daripada layanan terpisah (mis., Server penalaran (mis., Huggingface) atau menjalankan satu pekerja per pod. Secara umum, lebih banyak pod lebih besar daripada lebih banyak desain “lebih banyak pekerja per pod”, tetapi ada peringatan yang terkait dengan koneksi database yang aktif. Jika setiap pekerja menggunakan kumpulan koneksi, CPU dari CPU yang berkaitan dengan database yang aktif. (≈Pods × Workers × Pool_Size). Ini pada gilirannya menghasilkan latensi yang lebih tinggi, kesalahan koneksi, dan degradasi kinerja keseluruhan.

Tidak ada peluru perak. Pola dan praktik terbaik membantu, tetapi setiap solusi harus dibentuk oleh lingkungan bisnis tertentu.