在上篇文章我们学习了FastAPI教程04——路由分发，这篇我们学习FastAPI教程05——文件上传。

bytes类型

在FastAPI中，File用于处理文件上传，其语法格式如下：

参数变量名:bytes = File(参数)

参数和Path、Query、Field的参数差不多。

bytes类以二进制形式读取文件，将读取存储在内存中，示例代码如下：

import uvicorn
from fastapi import FastAPI, File
app = FastAPI()

@app.post("/upload")
async def upload_file(file: bytes = File(...)):
    with open('./file/img.jpg', 'wb') as f:
        f.write(file)
    return {"message": "文件上传成功"}

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True)

运行结果如下：

注意：bytes类适用于小于10MB的文件，内存占用高，大文件可能导致崩溃。

UploadFile类型

与bytes相比，UploadFile有更多的属性、方法，通过这些方法我们可以获取文件的元数据，读取文件内容。

主要有以下属性：

• filename：上传的文件名；

• content_type：文件的内容类型，如image/jpeg；

• file：SpooledTemporaryFile，文件对象；

• size：文件大小；

• headers：请求头信息；

主要有以下方法：

• read(size)：读取文件，size为1024的倍数；

• seek(指针)：文件指针移动到offset（int）字节位置；

• close()：关闭文件；

• write(data)：写入文件；

注意：所有方法都是async方法，所以都需要搭配await使用。

示例代码如下：

import uvicorn
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile):
    print('文件名：', file.filename)
    print("文件内容类型：", file.content_type)
    print("文件大小（字节）：", file.size)
    print("文件对象：", file.file)
    print("请求头信息：", file.headers)

    await file.read()  # 读取文件所有内容
    await file.seek(0)  # 文件指针重置到文件开头
    with open(f"./file/{file.filename}", "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 每次读取1m
            await f.write(chunk)  # 写入文件
    await file.close()      # 关闭文件
    return {"message": "文件上传成功"}

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True)

注意：代码中的with open是同步代码，遇到大文件时，可能会导致卡顿，这时我们可以使用aiofiles，首先执行如下代码安装aiofiles，

pip install aiofiles

接着把with open这行代码改为：

async with aiofiles.open(f"./file/{file.filename}", "wb") as f:

上传多个文件

当我们想上传多个文件时，可以搭配list实现，示例代码如下：

import uvicorn
from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload")
async def upload_file(files: list[UploadFile] = File(...)):
    print('文件数量：',len(files))
    for file in files:
        print("文件名：",file.filename)
    return {"message": "文件上传成功"}

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True)

运行结果如下：

上传到服务器

这里我们简单演示上传到腾讯云的对象存储中，示例代码如下：

import uvicorn
from fastapi import FastAPI, UploadFile, File
from qcloud_cos import CosS3Client
from qcloud_cos import CosConfig

config = CosConfig(
    Region='地区',   # 已创建桶归属的
    SecretId='SecretId',   # 账号密钥ID
    SecretKey='SecretKey',    # 账号密钥
    Token=None,     # 使用永久密钥不需要填
    Scheme='https'       # 指定使用 http/https 协议来访问 COS，默认为 https，可不填
)
client = CosS3Client(config)
app = FastAPI()


@app.post("/upload")
async def upload_file(files: UploadFile):
    chunk = await files.read()
    response = client.put_object(
        Bucket='桶名',
        Key=上传后的文件名,
        Body='字节'
    )
    url = client.get_presigned_download_url(
        Bucket='lin-1318425645',
        Key=files.filename,
        Expired=20  # 20秒有效期
    )
    # 永久链接
    url ='https://' + '桶名' + '.cos.' + 归属地 + '.myqcloud.com/' + 文件名
    return {"message": "文件上传成功"}

if __name__ == '__main__':
    uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True)

大家感兴趣可以通过腾讯云官方文档了解更多,，以后有机会会根据这个写一个详细版的上传文件到服务器。

好了，FastAPI教程05——文件上传就讲到这里了。

公众号：白巧克力LIN

该公众号发布Java、Python、数据库、Linux、Flask、Django、自动化测试、Git、算法、前端、服务器、AI等相关文章！

• END -