Mengapa FastAPI StreamingResponse saya tidak menstrim dengan fungsi penjana?

FastAPI StreamingResponse Tidak Menstrim dengan Fungsi Penjana

Masalah:

Aplikasi FastAPI gagal strim respons daripada fungsi penjana menggunakan StreamingResponse, menyebabkan keseluruhan respons dihantar secara keseluruhan.

Jawapan:

Terdapat beberapa faktor yang perlu dipertimbangkan semasa menggunakan StreamingResponse dengan fungsi penjana:

1. Jenis Permintaan HTTP:

Kod yang disediakan menggunakan permintaan POST, yang tidak sesuai untuk mendapatkan data daripada pelayan. Sebaliknya, gunakan permintaan GET untuk mengambil data.

2. Pengendalian Bukti Kelayakan:

Atas sebab keselamatan, elakkan menghantar bukti kelayakan (cth., 'auth_key') melalui rentetan pertanyaan URL. Gunakan pengepala atau kuki sebaliknya.

3. Sintaks Fungsi Penjana:

Operasi penyekatan tidak boleh dilaksanakan dalam fungsi penjana StreamingResponse. Gunakan def dan bukannya async def untuk fungsi penjana, kerana FastAPI menggunakan kumpulan benang untuk mengurus operasi penyekatan.

4. Penggunaan Iterator:

Dalam kod ujian anda, requests.iter_lines() berulang melalui data respons satu baris pada satu masa. Jika respons tidak mengandungi pemisah baris, gunakan iter_content() dan nyatakan saiz ketulan untuk mengelakkan masalah penimbalan yang berpotensi.

5. Jenis Media:

Pelayar boleh menimbal respons dengan media_type='text/plain'. Untuk mengelakkan ini, tetapkan media_type='text/event-stream' atau lumpuhkan MIME Sniffing menggunakan X-Content-Type-Options: nosniff dalam pengepala respons.

Contoh Berfungsi:

Berikut ialah contoh yang berfungsi dalam app.py dan test.py yang menangani isu yang dinyatakan di atas:

# app.py

from fastapi import FastAPI, StreamingResponse
import asyncio

app = FastAPI()

async def fake_data_streamer():
    for i in range(10):
        yield b'some fake data\n\n'
        await asyncio.sleep(0.5)

@app.get('/')
async def main():
    headers = {'X-Content-Type-Options': 'nosniff'}  # Disable MIME Sniffing
    return StreamingResponse(fake_data_streamer(), media_type='text/event-stream', headers=headers)

# test.py (using httpx)

import httpx

url = 'http://localhost:8000/'

with httpx.stream('GET', url) as r:
    for chunk in r.iter_content(1024):
        print(chunk)

Atas ialah kandungan terperinci Mengapa FastAPI StreamingResponse saya tidak menstrim dengan fungsi penjana?. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!