个人技术博客

0%

FastApi自定义响应之StreamingResponse、FileResponse

发表于 分类于

hello 大家好我是Monday，今天给大家带来一篇FastApi自定义响应之StreamingResponse、FileResponse的文章。

一、StreamingResponse

作用 ：采用异步生成器或普通生成器(generator)/迭代器(iterator)流式传输响应数据

实际代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

file_path = "test.mp4"
app = FastAPI()
@app.get("/")
def main():
    # 这是生成器函数。它是一个“生成器函数”，因为它里面包含了 yield 语句
    def iterfile():
        # 通过使用 with 块，确保在生成器函数完成后关闭类文件对象
        with open(file_path, "rb") as file_like:
            # yield from 告诉函数迭代名为 file_like 的东西
            # 对于迭代的每个部分，yield 的内容作为来自这个生成器函数
            yield from file_like
    return StreamingResponse(iterfile(), media_type="video/mp4"

二、FileResponse

作用：异步流式传输文件作为响应，重点一定是异步的

实际代码

1
2
3
4
5
6
7
8
9
10
from fastapi import FastAPI
from fastapi.responses import FileResponse

file_path = "test.mp4"
app = FastAPI()


@app.get("/file", response_class=FileResponse)
async def main():
    return file_path

三、FastApi利用FileResponse 做下载模块：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import datetime
from fastapi import FastAPI

import pandas
from starlette.responses import FileResponse

app = FastAPI()

@app.get("/download", summary="下载文件")
async def download_file():
    result = [
        {
            "id": 1,
            "name": "xxx",
            "age": 18
        }, {
            "id": 2,
            "name": "bbb",
            "age": 19
        }
    ]
    file = str(datetime.datetime.now().date())+".xlsx"
    df = pandas.DataFrame(result)
    df.columns = ["序号", "姓名", "年龄"]
    df.to_excel(file, index=False)
    return FileResponse(file, filename="user.xlsx")

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

问题: FileResponse文件下载接口响应后删除临时文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from fastapi import FastAPI
from starlette.responses import FileResponse
from starlette.background import BackgroundTasks


app = FastAPI()


def del_file(file_name):
    """
    下载完成后删除文件
    :param file_name:
    :return:
    """
    os.remove(file_name)

@app.get("/download")
def download():
    file_name = 'd:/test.csv'
    response = FileResponse(file_name, filename=file_name)
    task = BackgroundTasks()
    task.add_task(del_file, file_name)
    return response

四、FastApi利用StreamingResponse做下载模块：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import datetime
from fastapi import FastAPI
import io
import pandas
from starlette.responses import StreamingResponse

app = FastAPI()


@app.get("/download", summary="下载文件")
async def download_file():
    result = [
        {
            "id": 1,
            "name": "xxx",
            "age": 18
        }, {
            "id": 2,
            "name": "bbb",
            "age": 19
        }
    ]
    bio = io.BytesIO()
    df = pandas.DataFrame(result)
    writer = pandas.ExcelWriter(bio, engine='openpyxl')
    df.columns = ["序号", "姓名", "年龄"]
    df.to_excel(writer, sheet_name='数据列表', index=False)
    writer.save()
    bio.seek(0)
    file = str(datetime.datetime.now().date()) + ".xlsx"

    return StreamingResponse(bio, headers={
        'Content-Type': "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        'Cache-Control': "no-cache",
        'content-disposition': f"attachment; filename={file}"
    })


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8080)

参考链接：

https://blog.csdn.net/qq_33801641/article/details/120600652

结束语：

​ 今天的分享就到这里了，欢迎大家关注微信公众号”菜鸟童靴“