FastAPI 支持文件下载

news2025/6/8 22:26:58

FastAPI 支持文件下载
FastAPI 支持文件上传
Python 获取文件类型 mimetype

文章目录

  • 1. 服务端处理
    • 1.1. 下载小文件
    • 1.2. 下载大文件（yield 支持预览的）
    • 1.3. 下载大文件（bytes）
    • 1.4. 提供静态文件服务
  • 2. 客户端处理
    • 2.1. 普通下载
    • 2.2. 分块下载
    • 2.3. 显示进度条下载
    • 2.4. 带有断点续传的下载
    • 2.5. 带有超时和重试的下载
    • 2.6. 完整的下载器实现
  • 3. 异常处理
    • 3.1. 中文文件名错误
    • 3.2. 使用 ‌mimetypes.guess_type
    • 3.3. 使用 VUE Blob 下载乱码问题

参考：
https://blog.csdn.net/weixin_42502089/article/details/147689236
https://www.cnblogs.com/bitterteaer/p/17581746.html
修改下载缓冲区大小
https://ask.csdn.net/questions/8328950

对于文件下载，FastAPI 提供了 FileResponse 和 StreamingResponse 两种方式。 FileResponse 适合小文件，而 StreamingResponse 适合大文件，因为它可以分块返回文件内容。

1. 服务端处理

1.1. 下载小文件

使用 FileResponse 可以直接下载文件，而无需在内存中加载整个文件。

"""
fastapi + request 上传和下载功能
"""
from fastapi import FastAPI, UploadFile
from fastapi.responses import FileResponse
import uvicorn
 
app = FastAPI()

# filename 下载时设置的文件名
@app.get("/download/small/{filename}")
async def download_small_file(filename: str):
    print(filename)
    file_path = "./测试任务.pdf"
    return FileResponse(file_path, filename=filename, media_type="application/octet-stream")

if __name__ == '__main__':
    uvicorn.run(app, port=8000)

保证当前目录下有名为“测试任务.pdf”的文件。
然后使用浏览器下载：
http://127.0.0.1:8000/download/small/ceshi.pdf
在这里插入图片描述

1.2. 下载大文件（yield 支持预览的）

使用 StreamingResponse 可以分块下载文件，这样不会占用太多服务器资源，特别适用于大文件的下载。

from fastapi.responses import StreamingResponse
from fastapi import HTTPException
@app.get("/download/big/{filename}")
async def download_big_file(filename: str):
    def iter_file(path: str):
        with open(file=path, mode="rb") as tfile:
            yield tfile.read()

        # while chunk := tfile.read(1024*1024):  # 1MB 缓冲区
        #     yield chunk

    file_path = "./测试任务.pdf"
    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="File not found")
    
    # # 支持浏览器预览
    # return StreamingResponse(content=iter_file(path=file_path), status_code = 200,)
    # 直接下载
    return StreamingResponse(iter_file(path=file_path), media_type="application/octet-stream", 
                             headers={"Content-Disposition": f"attachment; filename={filename}"})

然后使用浏览器下载：
http://127.0.0.1:8000/download/big/ceshi_big.pdf

1.3. 下载大文件（bytes）

import io
@app.get("/download/bytes/{filename}")
async def download_bytes_file(filename: str):
    def read_bytes(path: str):
        content = "Error"
        with open(file=path, mode="rb") as tfile:
            content = tfile.read()
        # # 失败，需要转成bytes输出
        # return content
        return io.BytesIO(content)
        
    file_path = "./测试任务.pdf"
    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="File not found")

    # 解决中文名错误
    from urllib.parse import quote
    # return StreamingResponse(content=read_bytes(path=file_path), media_type="application/octet-stream", 
    #                          headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})
    return StreamingResponse(content=read_bytes(path=file_path), media_type="application/octet-stream", 
                             headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})

1.4. 提供静态文件服务

FastAPI 允许开发者使用 StaticFiles 来提供静态文件服务。这类似于传统 Web 服务器处理文件的方式。

from fastapi.staticfiles import StaticFiles

# app.mount("/static", StaticFiles(directory="static", html=True), name="free")
app.mount("/static", StaticFiles(directory="fonts", html=True), name="free")

尚未测试通过。

2. 客户端处理

参考（还有进度条， 带有断点续传的下载， 带有超时和重试的下载）：
https://blog.csdn.net/u013762572/article/details/145158401
批量上传下载
https://blog.csdn.net/weixin_43413871/article/details/137027968

2.1. 普通下载

import requests
import os

"""方式1，将整个文件下载在保存到本地"""
def download_file_bytes(file_name):
	# 以下三个地址均可以
    url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"
    url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
    url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"
    # response = requests.get(url, params={"filename": "1.txt"})
    response = requests.get(url)
    # print(response.text)
    with open(file_name, 'wb') as file:
        # file.write(response.text)
        file.write(response.content)

if __name__ == '__main__':
    download_file_bytes("本地测试下载文件bytes.pdf")

2.2. 分块下载

import requests
import os

"""方式2，通过流的方式一次写入8192字节"""
def download_file_big(file_name):
	# 以下三个地址均可以
    url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"
    # url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"
    # url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
    # response = requests.get(url, params={"filename": "./测试任务.pdf"}, stream=True)
    response = requests.get(url, stream=True)
    with open(file_name, 'wb') as file:
        for chunk in response.iter_content(chunk_size=8192):
            file.write(chunk)
 
if __name__ == '__main__':
    download_file_big("本地测试下载文件big.pdf")

2.3. 显示进度条下载

import requests
import os
from tqdm import tqdm

def download_file_tqdm(file_name):
	# 以下三个地址均可以
    # url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"
    # url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"
    url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
    response = requests.get(url, stream=True)
    if response.status_code == 200:
        file_size = int(response.headers.get('content-length', 0))
        
        # 显示进度条
        progress = tqdm(response.iter_content(chunk_size=8192), 
                       total=file_size,
                       unit='B', 
                       unit_scale=True)
        
        with open(file_name, 'wb') as f:
            for data in progress:
                f.write(data)
        return True
    return False

if __name__ == '__main__':
    download_file_tqdm("本地测试下载文件tqdm.pdf")

运行结果：

> python.exe .\fast_client.py
1.92kB [00:00, 14.0kB/s]

2.4. 带有断点续传的下载

# 带有断点续传的下载
def resume_download(file_name):
    # 以下三个地址均可以
    # url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"
    # url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"
    url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
    # 获取已下载文件大小
    initial_pos = os.path.getsize(file_name) if os.path.exists(file_name) else 0
    
    # 设置 Header
    headers = {'Range': f'bytes={initial_pos}-'}
    
    response = requests.get(url, stream=True, headers=headers)
    
    # 追加模式打开文件
    mode = 'ab' if initial_pos > 0 else 'wb'
    with open(file_name, mode) as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

尚未测试

2.5. 带有超时和重试的下载

# 带有超时和重试的下载
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import time
def download_with_retry(file_name, max_retries=3, timeout=30):
    # 以下三个地址均可以
    # url = "http://127.0.0.1:8000/download/small/ceshi_samll.pdf"
    # url = "http://127.0.0.1:8000/download/big/ceshi_big.pdf"
    url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
    session = requests.Session()
    
    # 设置重试策略
    retries = Retry(total=max_retries,
                   backoff_factor=1,
                   status_forcelist=[500, 502, 503, 504])
    
    session.mount('http://', HTTPAdapter(max_retries=retries))
    session.mount('https://', HTTPAdapter(max_retries=retries))
    
    try:
        response = session.get(url, stream=True, timeout=timeout)
        with open(file_name, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
        return True
    except Exception as e:
        print(f"Download failed: {str(e)}")
        return False

尚未测试

2.6. 完整的下载器实现

import requests
from tqdm import tqdm
import os
from pathlib import Path
import hashlib

class FileDownloader:
    def __init__(self, chunk_size=8192):
        self.chunk_size = chunk_size
        self.session = requests.Session()
        
    def get_file_size(self, url):
        response = self.session.head(url)
        return int(response.headers.get('content-length', 0))
    
    def get_file_hash(self, file_path):
        sha256_hash = hashlib.sha256()
        with open(file_path, "rb") as f:
            for byte_block in iter(lambda: f.read(4096), b""):
                sha256_hash.update(byte_block)
        return sha256_hash.hexdigest()
    
    def download(self, url, save_path, verify_hash=None):
        save_path = Path(save_path)
        # 创建目录
        save_path.parent.mkdir(parents=True, exist_ok=True)
        
        # 获取文件大小
        file_size = self.get_file_size(url)
        
        # 设置进度条
        progress = tqdm(total=file_size,
                       unit='B',
                       unit_scale=True,
                       desc=save_path.name)
        
        try:
            response = self.session.get(url, stream=True)
            with save_path.open('wb') as f:
                for chunk in response.iter_content(chunk_size=self.chunk_size):
                    if chunk:
                        f.write(chunk)
                        progress.update(len(chunk))
            progress.close()
            
            # 验证文件完整性
            if verify_hash:
                downloaded_hash = self.get_file_hash(save_path)
                if downloaded_hash != verify_hash:
                    raise ValueError("File hash verification failed")
                    
            return True
            
        except Exception as e:
            progress.close()
            print(f"Download failed: {str(e)}")
            if save_path.exists():
                save_path.unlink()
            return False
            
    def download_multiple(self, url_list, save_dir):
        results = []
        for url in url_list:
            filename = url.split('/')[-1]
            save_path = Path(save_dir) / filename
            success = self.download(url, save_path)
            results.append({
                'url': url,
                'success': success,
                'save_path': str(save_path)
            })
        return results

# 使用示例
downloader = FileDownloader()

# 单文件下载
url = "http://127.0.0.1:8000/download/bytes/ceshi_bytes.pdf"
downloader.download(url, save_path="downloads/file.pdf")

# # 多文件下载
# urls = [
#     "https://example.com/file1.pdf",
#     "https://example.com/file2.pdf"
# ]
# results = downloader.download_multiple(urls, "downloads")

运行结果：

> python.exe .\fast_client_plus.py
file.pdf: 9.18MB [00:00, 60.2MB/s]

3. 异常处理

3.1. 中文文件名错误

下载文件时，当传递文件名为中文时，报错。
在这里插入图片描述

    # 解决中文名错误
    from urllib.parse import quote
    return StreamingResponse(iter_file(path=file_path), media_type="application/octet-stream", 
                             headers={"Content-Disposition": "attachment; filename={}".format(quote(filename))})

在这里插入图片描述

3.2. 使用 ‌mimetypes.guess_type

mimetypes.guess_type‌ 是 Python 标准库 mimetypes 中的一个函数，用于根据文件名或 URL 猜测文件的 MIME 类型。该函数的基本用法如下：

import mimetypes
# 参数
#‌ 	url‌：可以是文件名或 URL 的字符串。
‌# 	strict‌：一个布尔值，默认为True，表示仅使用 IANA 注册的官方 MIME 类型。如果设置为 False，则还包括一些常用的非标准 MIM E类型。
# 返回值
#	函数返回一个元组 (type, encoding)：
‌#	type‌：文件的 MIME 类型，如果无法猜测则返回 None。
‌#	encoding‌：文件的编码格式，如果无法确定则返回 None。
mimetypes.guess_type(url, strict=True)

服务端修改 media_type

    from urllib.parse import quote
    from mimetypes import guess_type

    # 返回一个元组  ('application/pdf', None)
    content_type = guess_type(file_path)[0]

    return StreamingResponse(content=read_bytes(path=file_path), 
                             media_type=content_type, 
                             headers={
                                 "Content-Disposition": "attachment; filename={}".format(quote(filename))
                                 }
                            )

客户端返回：

{'date': 'Fri, 23 May 2025 01:04:55 GMT', 'server': 'uvicorn', 'content-disposition': 'attachment; filename=ceshi_bytes.pdf', 'content-type': 'application/pdf', 'Transfer-Encoding': 'chunked'}

3.3. 使用 VUE Blob 下载乱码问题

前端大佬说：
content-type 应该是 application/pdf
response-type 应该是 blob
在这里插入图片描述
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2386379.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

CNN卷积神经网络到底卷了啥？

CNN卷积神经网络到底卷了啥？

参考视频：卷积神经网络（CNN）到底卷了啥？8分钟带你快速了解！ 我们知道： 图片是由像素点构成，即最终的成像效果是由背后像素的颜色数值所决定 在Excel中：有这样一个由数值0和1组成的66…
阅读更多...
MIT 6.S081 2020Lab5 lazy page allocation 个人全流程

MIT 6.S081 2020Lab5 lazy page allocation 个人全流程

文章目录 零、写在前面一、Eliminate allocation from sbrk()1.1 说明1.2 实现 二、Lazy allocation2.1 说明2.2 实现 三、Lazytests and Usertests3.1 说明3.2 实现3.2.1 lazytests3.2.2 usertests 零、写在前面 可以阅读下4.6页面错误异常 像应用程序申请内存，内…
阅读更多...
C++初阶-list的使用2

C++初阶-list的使用2

目录 1.std::list::splice的使用 2.std::list::remove和std::list::remove_if的使用 2.1remove_if函数的简单介绍 基本用法 函数原型 使用函数对象作为谓词 使用普通函数作为谓词 注意事项 复杂对象示例 2.2remove与remove_if的简单使用 3.std::list::unique的使用 …
阅读更多...
如何优化 MySQL 存储过程的性能？

如何优化 MySQL 存储过程的性能？

文章目录 1. 优化 SQL 语句避免全表扫描减少子查询，改用 JOIN避免 SELECT 2. 合理使用索引3. 优化存储过程结构减少循环和临时变量避免重复计算 4. 使用临时表和缓存5. 优化事务处理6. 分析和监控性能7. 优化数据库配置8. 避免用户自定义函数（UDF&#…
阅读更多...
创新点！贝叶斯优化、CNN与LSTM结合，实现更准预测、更快效率、更高性能！

创新点！贝叶斯优化、CNN与LSTM结合，实现更准预测、更快效率、更高性能！

能源与环境领域的时空数据预测面临特征解析与参数调优双重挑战。CNN-LSTM成为突破口：CNN提取空间特征，LSTM捕捉时序依赖，实现时空数据的深度建模。但混合模型超参数（如卷积核数、LSTM层数）调优复杂，传统方法…
阅读更多...
【Flutter】创建BMI计算器应用并添加依赖和打包

【Flutter】创建BMI计算器应用并添加依赖和打包

😏★,:.☆(￣▽￣)/$:.★ 😏 这篇文章主要介绍创建BMI计算器应用并添加依赖和打包。 学其所用，用其所学。——梁启超 欢迎来到我的博客，一起学习，共同进步。 喜欢的朋友可以关注一下，下…
阅读更多...
【Linux 学习计划】-- 倒计时、进度条小程序

【Linux 学习计划】-- 倒计时、进度条小程序

目录 \r 、\n、fflush 倒计时 进度条 进度条进阶版 结语 \r 、\n、fflush 首先我们先来认识这三个东西，这将会是我们接下来两个小程序的重点之一 首先是我们的老演员\n，也就是回车加换行 这里面其实包含了两个操作，一个叫做回车&…
阅读更多...
微服务的应用案例

微服务的应用案例

从“菜市场”到“智慧超市”：一场微服务的变革之旅 曾经，我们的系统像一个熙熙攘攘的传统菜市场。所有功能模块（摊贩）都挤在一个巨大的单体应用中。用户请求（买菜的顾客）一多，整个市场就拥堵不堪…
阅读更多...
2025网络安全趋势报告 内容摘要

2025网络安全趋势报告 内容摘要

2025 年网络安全在技术、法规、行业等多个维度呈现新趋势。技术上，人工智能、隐私保护技术、区块链、量子安全技术等取得进展；法规方面，数据安全法规进一步细化；行业应用中，物联网、工业控制系统安全升级，供…
阅读更多...
云原生安全基石：深度解析HTTPS协议（从原理到实战）

云原生安全基石：深度解析HTTPS协议（从原理到实战）

🔥「炎码工坊」技术弹药已装填！ 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 一、基础概念：HTTPS是什么？ HTTPS（HyperText Transfer Protocol Secure）是HTTP协议的安全版本&#xff0c…
阅读更多...
Autodl训练Faster-RCNN网络--自己的数据集（一）

Autodl训练Faster-RCNN网络--自己的数据集（一）

参考文章： Autodl服务器中Faster-rcnn(jwyang)复现(一)_autodl faster rcnn-CSDN博客 Autodl服务器中Faster-rcnn(jwyang)训练自己数据集(二)_faster rcnn autodl-CSDN博客 食用指南：先跟着参考文章一进行操作，遇到问题再来看我这里有没有解…
阅读更多...
python打卡day36

python打卡day36

复习日 仔细回顾一下神经网络到目前的内容，没跟上进度的补一下进度 作业：对之前的信贷项目，利用神经网络训练下，尝试用到目前的知识点让代码更加规范和美观。探索性作业（随意完成）：尝试进入nn.M…
阅读更多...
Paimon和Hive相集成

Paimon和Hive相集成

Flink版本1.17 Hive版本3.1.3 1、Paimon集成Hive 将paimon-hive-connector.jar复制到auxlib中，下载链接Index of /groups/snapshots/org/apache/https://repository.apache.org/snapshots/org/apache/paimon/ 通过flink进入查看paimon /opt/softwares/flink-1.…
阅读更多...
HarmonyOS 鸿蒙应用开发进阶：深入理解鸿蒙跨设备互通机制

HarmonyOS 鸿蒙应用开发进阶：深入理解鸿蒙跨设备互通机制

鸿蒙跨设备互通（HarmonyOS Cross-Device Collaboration）是鸿蒙系统分布式能力的重要体现，通过创新的分布式软总线技术，实现了设备间的高效互联与能力共享。本文将系统性地解析鸿蒙跨设备互通的技术架构、实现原理及开发实践。 跨设…
阅读更多...
Vue.js教学第十五章：深入解析Webpack与Vue项目实战

Vue.js教学第十五章：深入解析Webpack与Vue项目实战

Webpack 与 Vue 项目详解 在现代前端开发中，Webpack 作为最流行的模块打包工具之一，对于 Vue 项目的构建和优化起着至关重要的作用。本文将深入剖析 Webpack 的基本概念、在 Vue 项目中的应用场景，并详细讲解常用的 Webpack loaders 和 plugins 的配置与作用，同时通过实例…
阅读更多...
Cmake编译gflags过程记录和在QT中测试

Cmake编译gflags过程记录和在QT中测试

由于在QT中使用PaddleOCR2.8存在这样那样的问题，查找貌似是gflags相关问题导致的，因此从头开始按相关参考文章编译一遍gflags源码，测试结果表明Qt5.14.2中使用MSVC2017X64编译器运行的QTgflags项目是正常。 详细编译步骤如下： 1、…
阅读更多...
项目中Warmup耗时高该如何操作处理

项目中Warmup耗时高该如何操作处理

1）项目中Warmup耗时高该如何操作处理 2）如何在卸载资源后Untracked和Other的内存都回收 3）总Triangles的值是否包含了通过GPU Instancing画的三角形 4）有没有用Lua来修复虚幻引擎中对C代码进行插桩Hook的方案 这是第432篇UWA技术知…
阅读更多...
制作一款打飞机游戏53：子弹样式

制作一款打飞机游戏53：子弹样式

现在，我们有一个小程序可以发射子弹，但这些子弹并不完美，我们稍后会修复它们。 子弹模式与目标 在开始之前，我想修正一下，因为我观察到在其他射击游戏中有一个我想复制的简单行为。我们有静态射击、瞄准射击和快速射击…
阅读更多...
Docker运维-5.3 配置私有仓库（Harbor）

Docker运维-5.3 配置私有仓库（Harbor）

1. harbor的介绍 Harbor(港湾)，是一个用于存储和分发 Docker 镜像的企业级 Registry 服务器。以前的镜像私有仓库采用官方的 Docker Registry，不便于管理镜像。 Harbor 是由 VMWare 在 Docker Registry 的基础之上进行了二次封装，加进去了很…
阅读更多...
day 36

day 36

利用前面所学知识，对之前的信贷项目，利用神经网络训练 # 先运行之前预处理好的代码 import pandas as pd import pandas as pd #用于数据处理和分析，可处理表格数据。 import numpy as np #用于数值计算，提供了高效的数组…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023