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

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

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

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

1.1 服務端 fastapi_server.py

編寫服務端代碼fastapi_server.py。

服務端代碼主要使用了fastapi和uvicorn兩個庫。

#!/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)  # 模擬每秒生成一個塊的耗時操作
        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編寫一個客戶端stream_client.py，進行流式接收。

#!/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/"  # 替換為你的實際接口地址


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中流式顯示

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

同時修改fastapi_server.py, 增加一個BlockIterator迭代器對長文本進行切塊。

<!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');

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

    // 使用 Fetch API 發(fā)送請求
    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;
                        }

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

</body>
</html>

服務端增加一個/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)》被譽為諸經(jīng)之首，大道之源，是中華優(yōu)秀傳統(tǒng)文化的總綱領，是中華民族五千年智慧的結(jié)晶。他含蓋萬有、綱紀群倫，是中華文化的杰出代表；他博大精微、包羅萬象，亦是中華文明的源頭。其內(nèi)容涉及哲學、生命、自然、科學、政治、天文、地理、文學、藝術等諸多領域，是各家共同的經(jīng)典。
《易經(jīng)》包括《連山》《歸藏》《周易》三部易書，現(xiàn)存于世的只有《周易》?！吨芤住废鄠魇侵芪耐醣磺袅h里時，研究《易經(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)  # 模擬每秒生成一個塊的耗時操作
        # 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)

總結(jié)

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章:

Pycharm激活碼激活兩種快速方式（附最新激活碼和插件）
這篇文章主要介紹了Pycharm兩種快速激活方式（附最新激活碼和插件） ,本文給大家分享兩種方式，小編在文章給大家推薦方法二，大家可以根據(jù)自己需要選擇，感興趣的朋友跟隨小編一起看看吧
2020-03-03
python3中利用filter函數(shù)輸出小于某個數(shù)的所有回文數(shù)實例
今天小編就為大家分享一篇 python3中利用filter函數(shù)輸出小于某個數(shù)的所有回文數(shù)實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-11-11
今天小編就為大家分享一篇python 使用poster模塊進行http方式的文件傳輸?shù)椒掌鞯姆椒?，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-01-01

Python生成器generator和yield關鍵字的使用

生成器是一種特殊的迭代器,可以通過列表推導式的修改或者使用yield關鍵字來創(chuàng)建,生成器函數(shù)能夠在迭代時動態(tài)產(chǎn)生值,而不是一次性生成所有值,這有助于節(jié)省內(nèi)存,yield關鍵字使函數(shù)執(zhí)行暫停并保存當前狀態(tài),下次調(diào)用時從停止處繼續(xù)執(zhí)行

2024-09-09