Fastapi之文件上传

1.小文件上传

1.单文件上传

1. import uvicorn
2. from fastapi import FastAPI
3. from fastapi import File
6. app = FastAPI()
9. @app.post("/file")
10. async def file_upload(file: bytes = File(..., max_length=2097152)):
11.     """使用File类，文件内容会以bytes的形式读入内存，适合小上传文件"""
12.     with open("D:\\lufei.jpg", "wb") as f:
13.         f.write(file)
14.     return {"file_size": len(file)}
17. if __name__ == '__main__':
18.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
19.                 debug=True, workers=1)

复制代码

2.多文件上传

1. import uvicorn
2. from fastapi import FastAPI
3. from fastapi import File
6. app = FastAPI()
9. @app.post("/file")
10. async def file_upload(file: List[bytes] = File(..., max_length=2097152)):
11.     """使用File类，文件内容会以bytes的形式读入内存，适合小上传文件"""
12.     return {"file_size": len(file)}
15. if __name__ == '__main__':
16.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
17.                 debug=True, workers=1)

复制代码

2.大文件上传

1.单文件上传

1. import uvicorn
2. from fastapi import FastAPI
3. from fastapi import UploadFile
6. app = FastAPI()
9. @users.post("/upload_file")
10. async def file_upload(file: UploadFile = File(...)):
11.     """
12.     使用UploadFile类的优势：
13.     1.文件开始存储在内存中，使用内存达到阈值后，将被保存在磁盘中
14.     2.适合于图片、视频大文件
15.     3.可以获取上传的文件的元数据，如文件名，创建时间等
16.     4.有文件对象的异步接口
17.     5.上传的文件是Python文件对象，可以使用write()、read()、seek()、close()等操做
18.     :param file:
19.     :return:
20.     """
22.     with open(f"D:\\{file.filename}", 'wb') as f:
23.         for i in iter(lambda: file.file.read(1024 * 1024 * 10), b''):
24.             f.write(i)
25.     f.close()
26.     return {"file_name": file.filename}
28. if __name__ == '__main__':
29.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
30.                 debug=True, workers=1)

复制代码

2.多文件上传

1. import uvicorn
2. from fastapi import FastAPI
3. from fastapi import UploadFile
6. app = FastAPI()
9. @users.post("/upload_file")
10. async def file_upload(files: List[UploadFile] = File(...)):
11.     for file in files:
12.         with open(f"D:\\{file.filename}", 'wb') as f:
13.             for i in iter(lambda: file.file.read(1024 * 1024 * 10), b''):
14.                 f.write(i)
15.         f.close()
16.     return {"file_name": [file.filename for file in files]}
19. if __name__ == '__main__':
20.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
21.                 debug=True, workers=1)

复制代码

3.大文件切片上传和断点续传

1.逻辑思路


2.代码示例

1. import os
2. import shutil
3. from pathlib import Path
5. from fastapi import FastAPI
6. import uvicorn
7. import aiofiles
8. from fastapi import Body, File, UploadFile
10. app = FastAPI()
12. base_dir = os.path.dirname(os.path.abspath(__file__))
13. upload_file_path = Path(base_dir, './uploads')
16. @app.post("/upload")
17. async def upload_file(
18.         identifier: str = Body(..., description="文件唯一标识符"),
19.         number: str = Body(..., description="文件分片序号（初值为0）"),
20.         file: UploadFile = File(..., description="文件")
21. ):
22.     """文件分片上传"""
23.     path = Path(upload_file_path, identifier)
24.     if not os.path.exists(path):
25.         os.makedirs(path)
26.     file_name = Path(path, f'{identifier}_{number}')
27.     if not os.path.exists(file_name):
28.         async with aiofiles.open(file_name, 'wb') as f:
29.             await f.write(await file.read())
30.     return {
31.         'code': 1,
32.         'chunk': f'{identifier}_{number}'
33.     }
36. @app.put("/merge")
37. async def merge_file(
38.         name: str = Body(..., description="文件名称（不含后缀）"),
39.         file_type: str = Body(..., description="文件类型/后缀"),
40.         identifier: str = Body(..., description="文件唯一标识符")
41. ):
42.     """合并分片文件"""
43.     target_file_name = Path(upload_file_path, f'{name}.{file_type}')
44.     path = Path(upload_file_path, identifier)
45.     try:
46.         async with aiofiles.open(target_file_name, 'wb+') as target_file:  # 打开目标文件
47.             for i in range(len(os.listdir(path))):
48.                 temp_file_name = Path(path, f'{identifier}_{i}')
49.                 async with aiofiles.open(temp_file_name, 'rb') as temp_file:  # 按序打开每个分片
50.                     data = await temp_file.read()
51.                     await target_file.write(data)  # 分片内容写入目标文件
52.     except Exception as e:
53.         return {
54.             'code': 0,
55.             'error': f'合并失败：{e}'
56.         }
57.     shutil.rmtree(path)  # 删除临时目录
58.     return {
59.         'code': 1,
60.         'name': f'{name}.{file_type}'
61.     }
64. if __name__ == '__main__':
65.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
66.                 debug=True, workers=1)

复制代码

4.大文件切片下载，断点续下

1.代码示例

1. import os
2. import re
3. import stat
4. from email.utils import formatdate
5. from mimetypes import guess_type
6. from pathlib import Path
7. from urllib.parse import quote
9. import uvicorn
10. from fastapi import FastAPI, Header
11. from fastapi import Path as F_Path
12. from starlette.responses import StreamingResponse
15. app = FastAPI()
17. base_dir = os.path.dirname(os.path.abspath(__file__))
18. upload_file_path = Path(base_dir, './uploads')
21. @app.get("/downloads/{file_name}")
22. async def download_file(
23.         scope: str = Header(default="bytes=0", regex=r"bytes=\d+"),
24.         file_name: str = F_Path(..., description="文件名称（含后缀）")
25. ):
26.     """分片下载文件，支持断点续传"""
27.     # 检查文件是否存在
28.     file_path = Path(upload_file_path, file_name)
29.     if not os.path.exists(file_path):
30.         return {
31.             'code': 0,
32.             'error': '文件不存在'
33.         }
34.     # 获取文件的信息
35.     stat_result = os.stat(file_path)
36.     content_type, encoding = guess_type(file_path)
37.     content_type = content_type or 'application/octet-stream'
38.     # 读取文件的起始位置和终止位置
39.     range_match = re.search(r'bytes=(\d+)-(\d+)', scope, re.S) or re.search(r'bytes=(\d+)-', scope, re.S)
40.     if range_match:
41.         start_bytes = int(range_match.group(1))
42.         end_bytes = int(range_match.group(2)) if range_match.lastindex == 2 else stat_result.st_size - 1
43.     else:
44.         start_bytes = 0
45.         end_bytes = stat_result.st_size - 1
46.     # 这里 content_length 表示剩余待传输的文件字节长度       判断是否为一般文件
47.     content_length = stat_result.st_size - start_bytes if stat.S_ISREG(stat_result.st_mode) else stat_result.st_size
48.     # 构建文件名称
49.     name, *suffix = file_name.rsplit('.', 1)
50.     suffix = f'.{suffix[0]}' if suffix else ''
51.     filename = quote(f'{name}{suffix}')  # 文件名编码，防止中文名报错
52.     # 打开文件从起始位置开始分片读取文件
53.     return StreamingResponse(
54.         file_iterator(file_path, start_bytes, 1024 * 1024 * 1),  # 每次读取 1M
55.         media_type=content_type,
56.         headers={
57.             'content-disposition': f'attachment; filename="{filename}"',
58.             'accept-ranges': 'bytes',
59.             'connection': 'keep-alive',
60.             'content-length': str(content_length),
61.             'content-range': f'bytes {start_bytes}-{end_bytes}/{stat_result.st_size}',
62.             'last-modified': formatdate(stat_result.st_mtime, usegmt=True),
63.         },
64.         status_code=206 if start_bytes > 0 else 200
65.     )
68. def file_iterator(file_path, offset, chunk_size):
69.     """
70.     文件生成器
71.     :param file_path: 文件绝对路径
72.     :param offset: 文件读取的起始位置
73.     :param chunk_size: 文件读取的块大小
74.     :return: yield
75.     """
76.     with open(file_path, 'rb') as f:
77.         f.seek(offset, os.SEEK_SET)
78.         while True:
79.             data = f.read(chunk_size)
80.             if data:
81.                 yield data
82.             else:
83.                 break
86. if __name__ == '__main__':
87.     uvicorn.run('main:app', host='0.0.0.0', port=8000, reload=True,
88.                 debug=True, workers=1)

复制代码

2.相关模块解读

• os.stat
  

1. import os
4. stat_result = os.stat("D:\lufei.jpg")
6. """
7. stat_result(
8.   st_mode=33206,   # 保护模式
9.   st_ino=844424930132018,  # 索引号
10.   st_dev=2222159220,  # 设备名
11.   st_nlink=1,  # 硬链接号（被连接数目）
12.   st_uid=0,  # 用户 ID
13.   st_gid=0,  # 组 ID
14.   st_size=20152,  # 文件大小，以字节为单位
15.   st_atime=1656044665,  # 最后一次访问时间
16.   st_mtime=1656038259,  # 文件最后一次修改时间
17.   st_ctime=1656038259  # 最后一次状态变化的时间，操作系统不会，该属性对应的结果也不同，例如在 Windows 操作系统下返回的就是文件的创建时间。
18. )
19. """

复制代码

• stat
  

1. import stat
3. if stat.S_ISREG(st_mode):           #判断是否一般文件
4.    print("xxxxx")
5. elif stat.S_ISLNK(st_mode):         #判断是否链接文件
6.    print("xxxxx")
7. elif stat.S_ISSOCK(st_mode):        #判断是否套接字文件   
8.    print("xxxxx")
9. elif stat.S_ISFIFO(st_mode):        #判断是否命名管道
10.    print("xxxxx")
11. elif stat.S_ISBLK(st_mode):         #判断是否块设备
12.    print("xxxxx")
13. elif stat.S_ISCHR(st_mode):         #判断是否字符设置
14. 　　print("xxxxx")
15. elif stat.S_ISDIR(st_mode):         #判断是否目录
16. 　　print("xxxxx")
18. # 查看文件权限
19. perm_code = stat.S_IMODE(st_mode)
20. perm = oct(perm_code)
22. # 获取文件类型
23. stat.S_IFMT(st_mode)

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！