✨✨ 欢迎大家来到景天科技苑✨✨

🎈🎈 养成好习惯，先赞后看哦~🎈🎈

🏆 作者简介：景天科技苑
🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。
🏆《博客》：Python全栈，PyQt5和Tkinter桌面开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi，flask等框架，云原生K8S，linux，shell脚本等实操经验，网站搭建，数据库等分享。

所属的专栏：Fastapi零基础入门与进阶实战教学
景天的主页：景天科技苑

FastAPI搭建文件下载和上传服务

引言

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API。它基于 Python 3.6+ 的类型提示，能够自动生成交互式 API 文档。在本教程中，我们将详细介绍如何使用 FastAPI 搭建一个文件下载服务。通过实际案例，我们将学习如何准备文件、创建路由、处理请求以及优化服务。

准备工作

安装必要的库

在开始之前，确保你的环境中安装了 Python 3.6 或更高版本，并安装了以下必要的库：

pip install fastapi uvicorn aiofiles pandas openpyxl

这些库的作用如下：

• fastapi：用于构建 Web API。
• uvicorn：用于运行 FastAPI 应用程序的 ASGI 服务器。
• aiofiles：用于异步文件操作。
• pandas 和 openpyxl：用于处理 Excel 文件。

准备Excel文件

我们将通过 pandas 创建一个 Excel 文件，并将其保存在项目中，以便下载。

import pandas as pd
from datetime import datetimedef prepare_excel_file():# 准备数据data = [{"id": 1, "name": "xxx", "age": 18},{"id": 2, "name": "bbb", "age": 19}]# 创建 DataFramedf = pd.DataFrame(data)df.columns = ["序号", "姓名", "年龄"]# 生成文件名file_name = str(datetime.now().date()) + ".xlsx"# 保存 Excel 文件df.to_excel(file_name, index=False)return file_name

这个函数会生成一个包含两行数据的 Excel 文件，文件名基于当前日期。

创建 FastAPI 应用

初始化 FastAPI

from fastapi import FastAPIapp = FastAPI()

文件下载路由

我们需要创建一个路由来处理文件下载请求。使用 FileResponse 来返回文件，这个类允许我们直接发送文件给客户端。

from starlette.responses import FileResponse@app.get("/download/excel")
async def download_excel():file_path = prepare_excel_file()  # 调用函数准备文件return FileResponse(file_path, filename="user_data.xlsx")

注意，这里我们使用了 prepare_excel_file 函数来准备 Excel 文件。然而，在实际应用中，你可能希望将文件保存在一个固定的位置，并在请求时直接读取这个文件，而不是每次请求都重新生成文件。

改进的文件下载

我们改进一下，将 Excel 文件预先生成并存放在项目中，然后直接返回这个文件。

# 假设文件已生成并存放在项目根目录下的 data/user_data.xlsx@app.get("/download/excel")
async def download_excel_improved():file_path = "data/user_data.xlsx"  # 文件路径return FileResponse(file_path, filename="user_data.xlsx")

运行 FastAPI 应用

使用 uvicorn 运行 FastAPI 应用：

uvicorn main:app --host 0.0.0.0 --port 8000

这里 main 是你的 Python 文件名（不包含 .py 后缀），app 是 FastAPI 实例的名称。

文件下载代码展示

import pandas as pd
from datetime import datetimeimport uvicorndef prepare_excel_file():# 准备数据data = [{"id": 1, "name": "xxx", "age": 18},{"id": 2, "name": "bbb", "age": 19}]# 创建 DataFramedf = pd.DataFrame(data)df.columns = ["序号", "姓名", "年龄"]# 生成文件名file_name = str(datetime.now().date()) + ".xlsx"# 保存 Excel 文件df.to_excel(file_name, index=False)return file_namefrom fastapi import FastAPIapp = FastAPI()from starlette.responses import FileResponse@app.get("/download/excel")
async def download_excel():file_path = prepare_excel_file()  # 调用函数准备文件#返回个文件对象，filename指定下载的文件名return FileResponse(file_path, filename="user_data.xlsx")if __name__ == '__main__':#注意，run的第一个参数 必须是文件名:应用程序名uvicorn.run("文件下载:app", port=8080,  reload=True)

浏览器访问下载地址：http://127.0.0.1:8080/download/excel
已将文件下载到本地

本地打开查看

拓展功能：文件上传和数据库管理

文件上传

FastAPI 支持文件上传，我们可以使用 UploadFile 类来处理上传的文件。

from fastapi import File, UploadFile
from typing import List@app.post("/upload/files/")
async def upload_files(files: List[UploadFile] = File(...)):uploaded_files = []for file in files:contents = await file.read()# 这里可以保存文件到服务器，或者进行其他处理uploaded_files.append({"filename": file.filename, "size": len(contents)})return {"message": "文件上传成功", "files": uploaded_files}

数据库管理

对于文件下载服务，如果文件存储在数据库中，我们可以根据文件ID或名称从数据库中检索文件信息，然后返回文件。

首先，需要设置数据库模型。这里以 SQLAlchemy 为例，创建一个简单的文件表。

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerBase = declarative_base()class FileModel(Base):__tablename__ = 'files'id = Column(Integer, primary_key=True)filename = Column(String(255), nullable=False)filepath = Column(String(512), nullable=False)# 配置数据库连接（这里以 SQLite 为例）
engine = create_engine('sqlite:///example.db', echo=True)
Base.metadata.create_all(engine)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 假设数据库中已有文件记录
# 接下来，我们编写一个根据文件ID下载文件的路由from fastapi import HTTPException@app.get("/download/file/{file_id}")
async def download_file_by_id(file_id: int, db: Session = SessionLocal()):# 从数据库中获取文件信息file_record = db.query(FileModel).filter(FileModel.id == file_id).first()if not file_record:raise HTTPException(status_code=404, detail="文件未找到")# 返回文件return FileResponse(path=file_record.filepath, filename=file_record.filename)# 注意：这里的 SessionLocal 是从 SQLAlchemy 会话工厂创建的，用于数据库操作。
# 你需要在 FastAPI 应用的依赖注入系统中注册它，以便在路由处理函数中自动传递数据库会话。
# 这里为了简化，我们没有展示依赖注入的完整设置。# 依赖注入的简单示例（通常放在 main.py 或其他配置文件中）
# from fastapi import Depends
# from sqlalchemy.orm import Session# def get_db(db_session: Session = Depends(SessionLocal)):
#     yield db_session# 然后，在路由函数中使用 Depends 注入数据库会话
# 但请注意，上面的 download_file_by_id 函数已经直接使用了 SessionLocal，这在实际应用中是不推荐的。
# 更好的做法是使用依赖注入来管理数据库会话的生命周期。

安全性和性能考虑

安全性

1. 验证和清理文件名：在返回文件时，确保文件名是安全的，防止路径遍历攻击（例如，通过 .. 访问上级目录）。
2. 权限验证：确保只有授权用户才能下载文件。
3. 文件类型限制：根据需求限制可上传和下载的文件类型。

性能

1. 缓存：对于频繁访问的文件，可以使用缓存来减少服务器负载和提高响应速度。
2. 异步处理：利用 FastAPI 和 aiofiles 的异步特性，处理文件读写和网络请求，提高并发处理能力。
3. 数据库优化：确保数据库索引正确，查询效率高效。

安全性与权限验证

在构建文件下载服务时，安全性是一个不可忽视的方面。特别是当服务涉及敏感文件或需要限制访问时，实现适当的权限验证至关重要。

权限验证基础

你可以使用 FastAPI 的依赖注入系统来管理权限验证逻辑。以下是一个简单的示例，展示了如何在下载文件之前验证用户的权限。

首先，定义一个模拟的用户验证函数（在实际应用中，这可能涉及检查数据库中的用户信息、JWT 令牌等）：

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBeareroauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")def get_current_user(token: str = Depends(oauth2_scheme)):# 这里只是一个模拟示例，实际中你应该根据token查询用户信息if token == "fake-super-secret-token":return {"username": "user123", "is_admin": True}else:raise HTTPException(status_code=403, detail="Invalid token")# 然后，在下载文件的路由中使用这个依赖
@app.get("/download/file/{file_id}")
async def download_file_with_auth(file_id: int, current_user: dict = Depends(get_current_user), db: Session = SessionLocal()):# 验证用户是否有权限下载文件（这里只是一个简单的示例）if not current_user.get("is_admin"):raise HTTPException(status_code=403, detail="You are not authorized to download this file")# 接下来是原有的文件下载逻辑...file_record = db.query(FileModel).filter(FileModel.id == file_id).first()if not file_record:raise HTTPException(status_code=404, detail="File not found")return FileResponse(path=file_record.filepath, filename=file_record.filename)

在这个示例中，get_current_user 函数通过检查一个模拟的令牌来验证用户身份。如果令牌有效，它将返回一个包含用户信息的字典；否则，将抛出一个 HTTPException 异常。然后，在 download_file_with_auth 路由中，我们通过 Depends 依赖项将 get_current_user 函数的结果作为 current_user 参数传入，并据此执行权限验证。

清理和验证文件名

当处理文件下载时，还需要确保文件名是安全的，以防止路径遍历攻击。你可以通过以下方式清理文件名：

1. 移除危险字符：使用正则表达式或其他字符串处理方法移除文件名中的 ..、/ 等可能导致路径遍历的字符。
2. 使用库函数：有些库提供了专门用于清理文件名的函数，可以利用这些库来避免手动处理可能出现的错误。

限制文件类型

根据需求，你可能还希望限制用户下载或上传的文件类型。这可以通过检查文件扩展名或MIME类型来实现。例如，在文件上传的路由中，你可以检查上传文件的MIME类型，确保它符合你的要求。

性能优化

在构建高性能的文件下载服务时，你可以考虑以下几个方面来优化性能：

1. 使用CDN：对于频繁访问的静态文件，使用内容分发网络（CDN）可以显著减少加载时间。
2. 缓存：在服务器端和客户端实施缓存策略，以减少对同一文件的重复请求。
3. 异步处理：利用 FastAPI 和 aiofiles 的异步特性来同时处理多个文件下载请求，提高并发处理能力。
4. 优化数据库查询：确保数据库索引正确，并使用高效的查询语句来检索文件信息。
5. 限制文件大小：如果可能，限制可下载文件的大小，以防止过大的文件对服务器造成不必要的负担。

通过结合使用这些技术和策略，你可以构建一个既安全又高效的FastAPI文件下载服务。

总结

通过本文的讲解，我们学习了如何使用 FastAPI 搭建一个简单的文件下载服务，包括文件准备、路由创建、文件上传和数据库管理。同时，我们也讨论了安全性和性能优化的相关考虑。FastAPI 凭借其高性能和易用性，是构建现代 Web 服务的理想选择。希望这个教程能帮助你更好地理解和使用 FastAPI 来构建你的文件下载服务。