FastAPI 支持文件下载
FastAPI 支持文件上传
Python 获取文件类型 mimetype
文章目录
1. 依赖库
前端上传文件文件一般都是使用 form-data,数据编码类型是 multipart/form-data,而不是 json 类型。如果你不是全属性安装 fastapi 那么需要单独的安装 form-data 相关的库:
# pip
pip install python-multipart
# poetry
poetry add python-multipart
2. 服务端处理
FastAPI 提供了 File() 和 UploadFile() 两个类用于处理文件上传。 File() 适用于小文件,而 UploadFile() 则更适合处理大文件。在文件上传时,我们建议使用异步版本的函数,这样可以避免阻塞服务器。
2.1. 上传小文件
使用 File() 时,可以通过 httpie 或 requests 库来模拟上传操作。需要注意的是,上传文件时应该使用表单而不是 JSON,这可以通过在命令中加入 -f 或 --form 参数来指定。
from fastapi import File
@app.post("/upload/small")
async def upload_small_file(small_file: bytes = File()) -> str:
return f"file size: {len(small_file)}"
2.2. 上传大文件
对于大文件,建议使用 UploadFile ,因为它会在服务器的磁盘上创建一个临时文件对象,而不是将整个文件加载到内存中。这样可以有效避免服务器内存溢出。
from fastapi import UploadFile
import time
import os
@app.post("/upload/big")
async def upload_big_file(big_file: UploadFile= File(...)):
filename = f"{str(time.time()).replace('.','')}-{big_file.filename}"
local_path = "upload"
local_file = os.path.join(local_path, filename)
if not os.path.exists(local_path):
os.mkdir(local_path)
with open(local_file, "wb") as f:
f.write(big_file.file.read())
f.flush()
return {
"filename": filename,
"filesize": big_file.size
}
2.3. 限制文件及错误处理
from fastapi import FastAPI, File, UploadFile, HTTPException
app = FastAPI()
MAX_FILE_SIZE = 1024 * 1024 * 10 # 10MB
ALLOWED_CONTENT_TYPES = ["image/jpeg", "image/png", "application/pdf"]
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
# 判断文件类型
if file.content_type not in ALLOWED_CONTENT_TYPES:
raise HTTPException(status_code=400, detail="不支持的文件类型")
# 判断文件大小
content = await file.read()
if len(content) > MAX_FILE_SIZE:
raise HTTPException(status_code=400, detail="文件大小超过限制(10MB)")
# 继续处理文件
return {"filename": file.filename, "message": "文件上传成功"}
限制文件类型:
定义支持文件类型列表,在文件上传之后,读取 content_type 属性获取文件类型,判断是否符合需求。然后抛出错误。
限制文件大小:
定义文件支持的大小 MAX_FILE_SIZE(一般使用配置文件),然后使用 file.read 方法读取文件 content,然后获取 len, 最后与 MAX_FILE_SIZE 进行对比,超出就抛出错误。
2.4. 多文件上传
FastAPI 支持同时上传多个文件。开发者可以通过声明一个包含 UploadFile 对象的列表来实现多文件上传。这里我们创建一个支持多种格式文件列表上传的 Demo!
"""
fastapi + request 上传和下载功能
"""
from fastapi import FastAPI, UploadFile
import uvicorn
app = FastAPI()
@app.post("/upload/files/")
async def create_upload_files(files: list[UploadFile]):
"""
处理多文件上传请求,返回上传文件的详细信息列表
"""
uploaded_files_info = []
for file in files:
# 读取文件内容
contents = await file.read()
# 统计文件内容的字节数
content_bytes = len(contents)
# 尝试解码文件内容为字符串,并统计字符数
try:
content_chars = len(contents.decode('utf-8'))
except UnicodeDecodeError:
content_chars = None
# 检查文件类型
if file.content_type.startswith("image/"):
file_content_info = "Binary image content"
elif content_chars is not None:
file_content_info = contents.decode('utf-8')
else:
file_content_info = "Binary content"
# 添加文件的详细信息到列表中
uploaded_files_info.append({
"filename": file.filename,
"content_type": file.content_type,
"size_in_bytes": content_bytes,
"size_in_chars": content_chars,
"file_content_info": file_content_info
})
# 返回所有上传文件的详细信息
return {"uploaded_files_info": uploaded_files_info}
if __name__ == '__main__':
uvicorn.run(app, port=8000)
2.5. 模板形式(单多文件)
如果是全局安装 poetry add fastapi[all] 那么,包里面就包含了 jinja 模板特性,直接用了。
单文件上传
from fastapi.responses import HTMLResponse
from fastapi import FastAPI, File, UploadFile, HTTPException
app = FastAPI()
@app.post("/file/")
async def create_file(file: UploadFile):
return file
@app.get("/")
async def main():
content = """
<body>
<form action="/file/" enctype="multipart/form-data" method="post">
<input name="file" type="file" multiple>
<input type="submit">
</form>
</body>
"""
return HTMLResponse(content=content)
if __name__ == '__main__':
uvicorn.run(app, port=8000)
测试结果:
浏览器输入:http://127.0.0.1:8000/
选择文件,提交。
多文件上传
有两个地方需要修改:
- 前端表单元素指定:multiple 属性
<input name="files" type="file" multiple>
- 后端使用 list 注解
async def create_files(files: list[bytes] = File()):
测试代码:
from fastapi.responses import HTMLResponse
from fastapi import FastAPI, File, UploadFile, HTTPException
app = FastAPI()
@app.post("/file/")
async def create_file(file: UploadFile):
return file
@app.post("/files/")
async def create_files(files: list[bytes] = File()):
return {"file_sizes": [len(file) for file in files]}
# @app.post("/uploadfiles/")
# async def create_upload_files(files: list[UploadFile]):
# return {"filenames": [file.filename for file in files]}
@app.get("/")
async def main():
content = """
<body>
<form action="/file/" enctype="multipart/form-data" method="post">
<input name="file" type="file" multiple>
<input type="submit">
</form>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
"""
return HTMLResponse(content=content)
if __name__ == '__main__':
uvicorn.run(app, port=8000)
测试结果:
选择两个文件
2.6. 其他说明
File 类
UploadFile 类
文件上传与表单
FastAPI 中可选性使用注解表示
前后端分离
额外 meta 信息
文件保存到本地
文件切片
以上参考:https://juejin.cn/post/7442707021282607143
3. 客户端处理
3.1. swagger 上传文件
3.2. postman 上传文件
3.2.1. 单文件上传
需要注意的是: form-data 的 KEY big_file 需要与接口中的参数名保持一致
3.2.2. 多文件上传
需要注意的是: form-data 的 KEY files 需要与接口中的参数名保持一致,且多个文件都是 files。
3.3. requests 上传文件
3.3.1. 单文件上传
import requests
import os
ADDRESS = "http://127.0.0.1:8000"
def upload_file_big(file_path):
url = f"{ADDRESS}/upload/big"
with open(file_path, 'rb') as f:
contents = f.read()
print(os.path.basename(file_path)) # example.pdf
response = requests.post(url, files={"big_file": (os.path.basename(file_path), contents, 'multipart/form-data')})
return response.json()
if __name__ == '__main__':
upload_file_big(r"./example.pdf")
3.3.2. 多文件上传
requests 库以 POST 方式上传多个文件,关键在于使用 files 参数.
以元组的形式列出要上传的文件,如 (‘files’, open(‘file1’, ‘rb’)) 和 (‘files’, open(‘file2’, ‘rb’))。
这种方式适用于后端接口参数名为 files 的多文件上传请求。
参考:https://blog.csdn.net/q389797999/article/details/120130559
import requests
import os
ADDRESS = "http://127.0.0.1:8000"
def upload_files(file_paths: list):
url = f"{ADDRESS}/upload/files"
# file_list = [
# ('files', ('image.jpg', open('./image.jpg', 'rb'), 'image/jpeg')),
# ('files', ('image2.jpg', open('./image2.jpg', 'rb'), 'image/jpeg'))
# ]
file_list = []
for file_path in file_paths:
if os.path.isfile(file_path):
file_key = os.path.basename(file_path)
file_list.append(("files", (file_key, open(file_path, 'rb'), 'image/jpeg')))
response = requests.post(url, files=file_list)
# 确保关闭文件
for file_item in file_list:
print(file_item[1])
file_item[1][1].close()
print(response.status_code)
print(response.json())
return response.json()
if __name__ == '__main__':
file_paths = ["image.jpg", "image2.jpg"]
upload_files(file_paths)
执行结果:
> python.exe .\fast_client.py
('image.jpg', <_io.BufferedReader name='image.jpg'>, 'image/jpeg')
('image2.jpg', <_io.BufferedReader name='image2.jpg'>, 'image/jpeg')
200
{'uploaded_files_info': [{'filename': 'image.jpg', 'content_type': 'image/jpeg', 'size_in_bytes': 321598, 'size_in_chars': None, 'file_content_info': 'Binary image content'}, {'filename': 'image2.jpg', 'content_type': 'image/jpeg', 'size_in_bytes': 64370, 'size_in_chars': None, 'file_content_info': 'Binary image content'}]}