如何使用Fastapi上传文件？先从请求体数据讲起

文章目录

1、请求体数据
2、form表单数据
3、小文件上传
- 1.单文件上传
- 2.多文件上传
4、大文件上传
- - 1.单文件上传
  - 2.多文件上传

1、请求体数据

前面我们讲到，get请求中，我们将请求数据放在url中，其实是非常不安全的，我们更愿意将请求数据放在请求体中。
当你需要将数据从客户端（例如浏览器）发送给 API 时，你将其作为「请求体」发送。请求体是客户端发送给 API 的数据。响应体是 API 发送给客户端的数据。

FastAPI 基于 Pydantic ，Pydantic 主要用来做类型强制检查（校验数据）。不符合类型要求就会抛出异常。

对于 API 服务，支持类型检查非常有用，会让服务更加健壮，也会加快开发速度，因为开发者再也不用自己写一行一行的做类型检查。

安装上手
pip install pydantic

from fastapi import FastAPI  # FastAPI 是一个为你的 API 提供了所有功能的 Python 类。
import uvicorn

from typing import Optional,Union,List

from pydantic import BaseModel,Field

from datetime import date

#创建应用程序，app是应用程序名
app = FastAPI()  # 这个实例将是创建你所有 API 的主要交互对象。这个 app 同样在如下命令中被 uvicorn 所引用


#fastapi要实现校验功能，需要借助pydantic这个模块，我们需要自己写个类，继承pydantic模块中的BaseModel，才能具有该功能
#在类型上做类型限制
class User(BaseModel):
    name:str = 'root'
    #默认是0，输入限制大于0，小于100
    age: int = Field(default=0, lt=100, gt=0)
    birth: Optional[date] = None
    #限制为数组，里面的元素限制为int类型
    friends: List[int] = []
    description: Union[str, None] = None




#异步的请求参数，函数加上async
@app.post("/data") #路径参数与查询参数共存
#传参data限制为User类型
async def data(data:User):
    #将查询结果返回
    return {}




if __name__ == '__main__':
    #注意，run的第一个参数 必须是文件名:应用程序名
    uvicorn.run("请求体数据:app", port=8080,  reload=True)

在docs测试，可以看到请求体限制数据类型
在这里插入图片描述

报错排查
在这里插入图片描述

报错：

TypeError: Failed to execute ‘fetch’ on ‘Window’: Request with GET/HEAD method cannot have body.

有@ResponseBody才会在接口中获取swagger列表

是由于方法中申明的是get方法却用了@requestBody

故将get 请求改为post 请求即可

当传参不符合限制要求，响应失败，提示年龄应小于100
在这里插入图片描述

当请求体参数完全符合要求，才能正确响应
在这里插入图片描述

我们可以将数据返回
#fastapi要实现校验功能，需要借助pydantic这个模块，我们需要自己写个类，继承pydantic模块中的BaseModel，才能具有该功能

#在类型上做类型限制
class User(BaseModel):
    name:str = 'root'
    #默认是0，输入限制大于0，小于100
    age: int = Field(default=0, lt=100, gt=0)
    birth: Optional[date] = None
    #限制为数组，里面的元素限制为int类型
    friends: List[int] = []
    description: Union[str, None] = None




#异步的请求参数，函数加上async
@app.post("/data") #路径参数与查询参数共存
#将传参data限制为User类型
async def data(data:User):
    print(data,type(data))
    #将查询结果返回
    return data

在这里插入图片描述

注意，当输入的数据类型跟限制类型不一致时，pydantic会尝试做数据类型转换，转换成功就可以正常返回，转换失败才报错

Field比较强大，可以做各种限制，甚至可以做正则限制 pattern

def Field(  # noqa: C901
    default: Any = PydanticUndefined,
    *,
    default_factory: typing.Callable[[], Any] | None = _Unset,
    alias: str | None = _Unset,
    alias_priority: int | None = _Unset,
    validation_alias: str | AliasPath | AliasChoices | None = _Unset,
    serialization_alias: str | None = _Unset,
    title: str | None = _Unset,
    description: str | None = _Unset,
    examples: list[Any] | None = _Unset,
    exclude: bool | None = _Unset,
    discriminator: str | types.Discriminator | None = _Unset,
    json_schema_extra: JsonDict | typing.Callable[[JsonDict], None] | None = _Unset,
    frozen: bool | None = _Unset,
    validate_default: bool | None = _Unset,
    repr: bool = _Unset,
    init: bool | None = _Unset,
    init_var: bool | None = _Unset,
    kw_only: bool | None = _Unset,
    pattern: str | None = _Unset,
    strict: bool | None = _Unset,
    gt: float | None = _Unset,
    ge: float | None = _Unset,
    lt: float | None = _Unset,
    le: float | None = _Unset,
    multiple_of: float | None = _Unset,
    allow_inf_nan: bool | None = _Unset,
    max_digits: int | None = _Unset,
    decimal_places: int | None = _Unset,
    min_length: int | None = _Unset,
    max_length: int | None = _Unset,
    union_mode: Literal['smart', 'left_to_right'] = _Unset,
    **extra: Unpack[_EmptyKwargs],

也可以自定义一个函数做限制，使用到了pydantic里面的validator装饰器
最新版的validator已被废弃
在这里插入图片描述

最新版使用field_validator装饰器

#在类型上做类型限制
class User(BaseModel):
    name:str = 'root'
    #默认是0，输入限制大于0，小于100
    age: int = Field(default=0, lt=100, gt=0)
    birth: Optional[date] = None
    #限制为数组，里面的元素限制为int类型
    friends: List[int] = []
    description: Union[str, None] = None

    @field_validator('name')
    def validate_name(cls,v):
        assert v.isalpha(), 'name must be alpha'
        return v

校验生效
在这里插入图片描述

类型嵌套：
我们定义的类型，可以组合嵌套方式使用
在这里插入图片描述

class Data(BaseModel): # 类型嵌套
users: List[User]

@app.post(“/data/”)
async def create_data(data: Data):
# 添加数据库
return data

也可以这样嵌套，请求体数据是列表套字典形式

2、form表单数据

在 OAuth2 规范的一种使用方式（密码流）中，需要将用户名、密码作为表单字段发送，而不是 JSON。

FastAPI 可以使用Form组件来接收表单数据，需要先使用 pip install python-multipart 命令进行安装。
pip install python-multipart
在这里插入图片描述

from fastapi import FastAPI, Form
import uvicorn


app = FastAPI()

@app.post("/regin")
def regin(username: str = Form(..., max_length=16, min_length=8, pattern='[a-zA-Z]'),   #Form对输入的数据可以做些限制
          password: str = Form(..., max_length=16, min_length=8, pattern='[0-9]')):
    print(f"username:{username},password:{password}")
    return {"username": username}




if __name__ == '__main__':
    #注意，run的第一个参数 必须是文件名:应用程序名
    uvicorn.run("表单:app", port=8080,  reload=True)

在这里插入图片描述

此时发送请求，content-type 必须是application/x-www-form-urlencoded
否则发送请求失败
在这里插入图片描述

使用application/x-www-form-urlencoded发送成功
在这里插入图片描述

3、小文件上传

文件上传，文件会放在请求体里面，但是请求头的content-type是multipart/form-data

1.单文件上传

# file: bytes = File()：适合小文件上传
@app.post("/files/")
#文件时字节流类型，是fastapi里面的File类型
async def create_file(file: bytes = File()):
    print("file:", file)
    return {"file_size": len(file)}

在docs请求测试，可以看到请求的content-type是multipart/form-data
返回了图片的字节流长度
在这里插入图片描述

看下后台打印
在这里插入图片描述

但是这样上传只适合小文件，因为上传的文件会占用用户内存，太大的话会把内存撑爆

2.多文件上传

#多文件上传
@app.post("/multiFiles/")
async def create_files(files: List[bytes] = File()):
    for file in files:
        print(len(file))
    return {"file_sizes": [len(file) for file in files]}

点一次Add string item，就会增加一个文件上传按钮
在这里插入图片描述

看下后台打印
在这里插入图片描述

4、大文件上传

文件比较大时，如果一次性上传，可能会把用户内存撑爆，因此比较常见的处理方式就是分批上传。
上传大文件使用fastapi的UploadFile

1.单文件上传

from fastapi import FastAPI, File, UploadFile

# file: UploadFile：适合大文件上传，比较常用

@app.post("/uploadFile/")
#直接对应UploadFile类型数据
async def create_upload_file(file: UploadFile):
    #打印文件名称
    print('file',file.filename)
    #将上传的文件保存到服务本地
    with open(f"{file.filename}", 'wb') as f:
        #一次读取1024字节，循环读取写入
        for chunk in iter(lambda: file.file.read(1024), b''):
            f.write(chunk)

    return {"filename": file.filename}

在这里插入图片描述

后台打印
在这里插入图片描述

可以看到上传的文件被保存在服务端本地
在这里插入图片描述

单文件上传完整代码：

from fastapi import FastAPI, File, UploadFile
from typing import List
import uvicorn

app = FastAPI()

# file: UploadFile：适合大文件上传，比较常用

@app.post("/uploadFile/")
#直接对应UploadFile类型数据
async def create_upload_file(file: UploadFile):
    #打印文件名称
    print('file',file.filename)
    #将上传的文件保存到服务本地
    with open(f"{file.filename}", 'wb') as f:
        #一次读取1024字节，循环读取写入
        for chunk in iter(lambda: file.file.read(1024), b''):
            f.write(chunk)

    return {"filename": file.filename}


if __name__ == '__main__':
    #注意，run的第一个参数 必须是文件名:应用程序名
    uvicorn.run("文件上传:app", port=8080,  reload=True)

2.多文件上传

#上传多个文件
@app.post("/multiUploadFiles/")
async def create_upload_files(files: List[UploadFile]):
    for file in files:
        print(file.filename)
        # 将上传的文件保存到服务本地
        path = os.path.join('images',f'{file.filename}')
        with open(path, 'wb') as f:
            # 一次读取1024字节，循环读取写入
            for chunk in iter(lambda: file.file.read(1024), b''):
                f.write(chunk)

    return {"filenames": [file.filename for file in files]}

在这里插入图片描述

看下后台打印，以及上传的文件
在这里插入图片描述

查看下载的文件
在这里插入图片描述

多文件上传代码：

from fastapi import FastAPI, File, UploadFile
from typing import List
import uvicorn
import os

app = FastAPI()


#上传多个文件
@app.post("/multiUploadFiles/")
async def create_upload_files(files: List[UploadFile]):
    for file in files:
        print(file.filename)
        # 将上传的文件保存到服务本地
        path = os.path.join('images',f'{file.filename}')
        with open(path, 'wb') as f:
            # 一次读取1024字节，循环读取写入
            for chunk in iter(lambda: file.file.read(1024), b''):
                f.write(chunk)

    return {"filenames": [file.filename for file in files]}



if __name__ == '__main__':
    #注意，run的第一个参数 必须是文件名:应用程序名
    uvicorn.run("文件上传:app", port=8080,  reload=True)