Python使用fastAPI如何实现一个流式传输接口
更新时间：2024年06月24日 08:50:51 作者：babybin
这篇文章主要介绍了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 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参数为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 发送请求
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 uvicorn
app = FastAPI()
CONTENT = """《易经》被誉为诸经之首，大道之源，是中华优秀传统文化的总纲领，是中华民族五千年智慧的结晶。他含盖万有、纲纪群伦，是中华文化的杰出代表；他博大精微、包罗万象，亦是中华文明的源头。其内容涉及哲学、生命、自然、科学、政治、天文、地理、文学、艺术等诸多领域，是各家共同的经典。
《易经》包括《连山》《归藏》《周易》三部易书，现存于世的只有《周易》。《周易》相传是周文王被囚羑里时，研究《易经》所作的结论。"""
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)
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。
您可能感兴趣的文章:
相关文章
这篇文章主要介绍了Python3逻辑运算符与成员运算符，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的朋友可以参考一下2022-09-09
在 Python 中处理电子邮件时,标准库中的 email 模块是首选工具,无论你需要发送 HTML 格式的邮件、带附件的邮件,还是解析复杂的邮件结构,email 库都能胜任,这篇博客将带你系统地认识并掌握 email 模块的使用方法与内部机制,需要的朋友可以参考下2025-06-06
今天小编就为大家分享一篇python去掉 unicode 字符串前面的u方法。具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2018-10-10
这篇文章主要介绍了python Yaml 、Json 、Dict 之间的转化的示例，帮助大家更好的理解和学习python，感兴趣的朋友可以了解下2020-10-10
这篇文章主要介绍了Python中的DateTime和TimeDelta详解,在Python中，date，time和datetime类提供了许多函数来处理日期、时间和时间间隔,每当您操纵日期或时间时，都需要导入DateTime函数,需要的朋友可以参考下2023-07-07
这篇文章主要介绍了python检测远程udp端口是否打开的方法,涉及Python操作socket实现检测udp端口的技巧,需要的朋友可以参考下2015-03-03
今天小编就为大家分享一篇Pycharm之快速定位到某行快捷键的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2019-01-01
这篇文章主要介绍了使用django-crontab实现定时任务,小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-02-02
今天小编就为大家分享一篇pytorch实现mnist数据集的图像可视化及保存，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-01-01
这篇文章主要为大家详细介绍了Python办公自动化中数据预处理和数据校验的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以参考一下2024-01-01
最新评论