Python使用fastAPI实现一个流式传输接口-编程知识

Python使用fastAPI实现一个流式传输接口

1. 使用fastapi实现流式传输

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 uvicornapp = 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.pyimport requestsurl = "http://127.0.0.1:8001/stream/"  # 替换为你的实际接口地址def test1():try:response = requests.get(url, stream=True) # stream参数为Trueif 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 发送请求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;}// 将每个块的内容添加到页面上outputDiv.innerHTML += new TextDecoder().decode(value) +"<br>";}}});}).then(stream => {// 使用 TextStream，将流连接到页面上的输出const textStream = new TextStream(stream);return textStream.pipeTo(new WritableStream({write: chunk => {// 在这里你可以处理每个块的数据console.log('Received chunk:', chunk);}}));}).catch(error => console.error('Error:', error));
</script></body>
</html>

服务端增加一个/web_client接口，读取上述网页内容

#!/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 uvicornapp = FastAPI()CONTENT = """《易经》被誉为诸经之首，大道之源，是中华优秀传统文化的总纲领，是中华民族五千年智慧的结晶。他含盖万有、纲纪群伦，是中华文化的杰出代表；他博大精微、包罗万象，亦是中华文明的源头。其内容涉及哲学、生命、自然、科学、政治、天文、地理、文学、艺术等诸多领域，是各家共同的经典。
《易经》包括《连山》《归藏》《周易》三部易书，现存于世的只有《周易》。《周易》相传是周文王被囚羑里时，研究《易经》所作的结论。"""class BlockIterator:def __init__(self, text, block_size=10):self.text = textself.block_size = block_sizeself.index = 0def __iter__(self):return selfdef __next__(self):if self.index >= len(self.text):raise StopIterationblock = self.text[self.index:self.index + self.block_size]self.index += self.block_sizereturn blockasync 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)