Python使用fastAPI如何實(shí)現(xiàn)一個(gè)流式傳輸接口

更新時(shí)間：2024年06月24日 08:50:51 作者：babybin

這篇文章主要介紹了Python使用fastAPI如何實(shí)現(xiàn)一個(gè)流式傳輸接口問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

1. 使用fastapi實(shí)現(xiàn)流式傳輸

1.1 服務(wù)端 fastapi_server.py

編寫服務(wù)端代碼fastapi_server.py。

服務(wù)端代碼主要使用了fastapi和uvicorn兩個(gè)庫。

#!/usr/bin/env python
# coding=utf-8
# @Time    : 2024/1/31 19:13
# @Software: PyCharm
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time
import uvicorn

app = FastAPI()


async def generate_data():
    for i in range(1, 11):
        time.sleep(1)  # 模擬每秒生成一個(gè)塊的耗時(shí)操作
        yield f"FASTAPI Chunk {i}\n"


@app.get("/stream")
async def stream_data():
    return StreamingResponse(generate_data(), media_type="text/plain")


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

1.2客戶端（requests）

使用python編寫一個(gè)客戶端stream_client.py，進(jìn)行流式接收。

#!/usr/bin/env python
# coding=utf-8
# @Time    : 2024/1/31 19:14
#
# stream_client.py


import requests

url = "http://127.0.0.1:8001/stream/"  # 替換為你的實(shí)際接口地址


def test1():
    try:
        response = requests.get(url, stream=True) # stream參數(shù)為True

        if response.status_code == 200:
            for chunk in response.iter_content(chunk_size=7):  # 這行很重要哦
                if chunk:
                    print(chunk.decode("utf-8"), end="")
    except requests.RequestException as e:
        print(f"Request failed: {e}")


def test2():
    try:
        response = requests.get(url, stream=True)

        if response.status_code == 200:
            for line in response.iter_lines(decode_unicode=True, chunk_size=8):
                if line:
                    print("Received SSE event:", line)
    except requests.RequestException as e:
        print(f"Request failed: {e}")


# test1()
test2()

1.3 在html中流式顯示

新建一個(gè)client.html文件，放在fastapi_server.py目錄下。

同時(shí)修改fastapi_server.py, 增加一個(gè)BlockIterator迭代器對(duì)長文本進(jìn)行切塊。

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Streaming Client (client.html)</title>
</head>
<body>
aaa
<div id="output"></div>

<script>
    const outputDiv = document.getElementById('output');

    // 替換為你的實(shí)際接口地址
    const url = '$FASTAPI$';

    // 使用 Fetch API 發(fā)送請(qǐng)求
    fetch(url)
        .then(response => {
            const reader = response.body.getReader();
            return new ReadableStream({
                async start(controller) {
                    while (true) {
                        const { done, value } = await reader.read();

                        // 如果讀取完畢，中止流
                        if (done) {
                            controller.close();
                            break;
                        }

                        // 將每個(gè)塊的內(nèi)容添加到頁面上
                        outputDiv.innerHTML += new TextDecoder().decode(value) +"<br>";
                    }
                }
            });
        })
        .then(stream => {
            // 使用 TextStream，將流連接到頁面上的輸出
            const textStream = new TextStream(stream);
            return textStream.pipeTo(new WritableStream({
                write: chunk => {
                    // 在這里你可以處理每個(gè)塊的數(shù)據(jù)
                    console.log('Received chunk:', chunk);
                }
            }));
        })
        .catch(error => console.error('Error:', error));
</script>

</body>
</html>

服務(wù)端增加一個(gè)/web_client接口，讀取上述網(wǎng)頁內(nèi)容

#!/usr/bin/env python
# coding=utf-8
# @Time    : 2024/1/31 19:13
# @Software: PyCharm
from fastapi import FastAPI
from starlette.responses import HTMLResponse
from fastapi.responses import StreamingResponse
import time
import uvicorn

app = FastAPI()

CONTENT = """《易經(jīng)》被譽(yù)為諸經(jīng)之首，大道之源，是中華優(yōu)秀傳統(tǒng)文化的總綱領(lǐng)，是中華民族五千年智慧的結(jié)晶。他含蓋萬有、綱紀(jì)群倫，是中華文化的杰出代表；他博大精微、包羅萬象，亦是中華文明的源頭。其內(nèi)容涉及哲學(xué)、生命、自然、科學(xué)、政治、天文、地理、文學(xué)、藝術(shù)等諸多領(lǐng)域，是各家共同的經(jīng)典。
《易經(jīng)》包括《連山》《歸藏》《周易》三部易書，現(xiàn)存于世的只有《周易》?！吨芤住废鄠魇侵芪耐醣磺袅h里時(shí)，研究《易經(jīng)》所作的結(jié)論。"""


class BlockIterator:
    def __init__(self, text, block_size=10):
        self.text = text
        self.block_size = block_size
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.text):
            raise StopIteration

        block = self.text[self.index:self.index + self.block_size]
        self.index += self.block_size
        return block


async def generate_data():
    for i in BlockIterator(CONTENT, block_size=1):
        time.sleep(0.05)  # 模擬每秒生成一個(gè)塊的耗時(shí)操作
        # yield f"FASTAPI Chunk {i}\n"
        yield i


@app.get("/web_client", response_class=HTMLResponse)
async def read_root():
    with open("static/client.html", "r") as file:
        html_content = file.read()
        html_content = html_content.replace("$FASTAPI$", "http://127.0.0.1:8001/stream/")
    return HTMLResponse(content=html_content)


@app.get("/stream")
async def stream_data():
    return StreamingResponse(generate_data(), media_type="text/plain")


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)