处理HTTP请求

这篇笔记我们介绍如何在FastAPI中处理HTTP请求。

定义处理函数

还是以最开始的Hello, world!为例，这里定义了一个名为root的处理函数，它匹配/路径。

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def root():
    return {'message': 'Hello, world!'}

我们可以看到root函数使用async def定义，它是一个异步函数，如果我们函数内部的代码逻辑涉及许多可以异步进行的操作，我们应当尽量基于Python的原生异步并发编程模型来开发，这样我们的代码通常会具有更好的并发性能。

当然，如果我们使用的某些库没有异步替代，在FastAPI中我们仍可以使用普通函数作为路由处理函数，FastAPI框架会自动将其放入线程池执行。

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
def root():
    return {'message': 'Hello, world!'}

注意：不要在异步函数中使用同步阻塞操作，这样会阻塞事件循环，导致并发性能急剧下降。

接收HTTP请求

处理请求URL参数

下面例子中，我们编写了一个获取GET查询参数的例子。

from fastapi import FastAPI

app = FastAPI()


@app.get('/queryUser')
async def query_user(user_code: int, username: str | None):
    return {'user_code': user_code, 'username': username}

代码中，我们要求/queryUser接收一个必选的int类型参数user_code，此外还有一个可选的str类型参数username。它们都是强类型的，必选参数和可选参数通过形如str | None的写法来控制，当类型不符合时将会抛出相应的错误。

我们可以访问http://localhost:8000/queryUser?user_code=1&username=abc来验证输出结果。

此外还有一点我们需要注意。Python语言编写Web接口时，一个需要额外考虑的问题是Python的编码风格使用SnakeCase，但许多开发规范中Web接口的参数使用CamelCase。对于GET参数，我们可以使用FastAPI的查询参数别名来实现。

from fastapi import FastAPI, Query

app = FastAPI()


@app.get('/queryUser')
async def query_user(username: str | None, user_code: int = Query(alias='userCode')):
    return {'user_code': user_code, 'username': username}

上面代码中，user_code: int = Query(None, alias='userCode')是一个函数的参数定义，参数名为user_code，类型为int，等号后边的是参数的默认值，Query是FastAPI提供的用于获取查询参数的函数，它定义了参数的别名，我们这里将其指定为userCode。此时，我们就可以使用userCode来代替user_code作为GET参数名了。

补充：

对于可选参数，我们可以在Query中指定默认值，例如Query(None, alias='userCode')
alias支持传入一个数组，指定多个别名

处理Restful风格的路径参数

关于路径参数前面我们已经介绍过了，下面是一个例子。

from fastapi import FastAPI

app = FastAPI()


@app.get('/user/{user_id:str}')
async def root(user_id: str):
    return {'message': user_id}

读取JSON请求体

读取JSON请求体的最基础方式是使用一个dict类型的参数接收，下面是一个例子。

from fastapi import FastAPI

app = FastAPI()


@app.post('/user')
async def add_user(user: dict):
    return {'user': user}

不过实际开发中，除非请求体特别简单，一般不推荐这样编写。因为这样的代码可读性很差，仅从后端代码我们几乎很难看出user具体有哪些字段，一个比较好的习惯是将所有的请求数据封装到一个类中，下面是一个例子。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    username: str
    user_code: int


@app.post('/user')
async def add_user(user: User):
    return {'user': user}

FastAPI中数据模型的定义是基于pydantic库的，我们定义的User类继承BaseModel，其中包含了我们所需的字段和其对应的类型。

当然，这里也存在CamelCase的转换问题，如果我们需要驼峰命名的模型，我们同样可以使用别名解决。

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class User(BaseModel):
    username: str
    user_code: int = Field(alias='userCode')


@app.post('/user')
async def add_user(user: User):
    return {'user': user}

Field是pydantic提供的类，用于对字段进行额外信息的指定，这里我们使用alias指定了别名。此时这个模型在接收和输出JSON数据时，都会采用这个别名。

接收表单请求

如果请求体类型是表单，底层会使用python-multipart库进行处理，不过安装FastAPI时通常这个库也会自动安装。对于x-www-form-urlencoded和form-data类型的表单写法都是一致的。

from fastapi import FastAPI, Form
from pydantic import BaseModel, Field

app = FastAPI()


class User(BaseModel):
    username: str
    user_code: int = Field(alias='userCode')


@app.post('/user')
async def add_user(username: str = Form(), user_code: int = Form()):
    return {'user': {
        'username': username,
        'user_code': user_code
    }}

表单文件上传同样由python-multipart库进行处理，下面是一个例子。

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post('/upload')
async def upload(file: UploadFile = File()):
    content = await file.read()
    content_str = content.decode('utf-8')
    return {'size': file.size, 'filename': file.filename, 'content': content_str}

文件上传需要使用UploadFile类型，其中包含了文件大小、文件名、内容等信息，上面代码我们直接将所有数据以字符串形式一次性读取了出来，实际开发中，对于大文件我们可能需要分块写入磁盘。

关于表单字段的CamelCase问题，解决方案也是和GET参数一样的，我们在对应的Form()或File()中指定别名就行了。

读取请求头信息

读取请求头需要用到Header对象，下面是一个例子。

from fastapi import FastAPI, Header

app = FastAPI()


@app.get('/ua')
async def get_ua(user_agent: str | None = Header(default=None)):
    return user_agent

代码中，我们读取了User-Agent这个请求头，如果请求头不存在则返回None。

读取Cookie信息

读取Cookie信息需要用到Cookie对象，下面是一个例子。

from fastapi import FastAPI, Cookie

app = FastAPI()


@app.get('/cookie')
async def get_cookie(user_id: str | None = Cookie(default=None, alias='userId')):
    return user_id

代码中，我们使用别名方式读取了userId这个Cookie。

返回HTTP响应

返回文本数据

返回HTTP响应最基础的方式是使用Response对象，下面是一个例子。

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()


@app.get('/foo')
async def foo():
    return Response(content='hello')

例子中，我们创建了Response对象，其中的内容是一些文本。注意，当你直接返回Response时，它的数据既没有校验，又不会进行序列化，也不会自动生成文档，实际开发中，对于普通的HTTP接口，我们倾向于返回数据模型对象。

返回JSON数据

返回JSON最基础的方式是返回一个dict对象。

from fastapi import FastAPI

app = FastAPI()


@app.get('/user')
async def get_user():
    return {
        'username': 'Tom',
        'userCode': '1123'
    }

不过实际开发中，我们应该尽量将数据封装成类，FastAPI中我们可以返回基于pydantic的数据模型类。

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class User(BaseModel):
    username: str
    user_code: int = Field(alias='userCode')


@app.get('/user')
async def get_user():
    user = User(username='Tom', userCode='123')
    return user

此外，我们还可以使用JSONResponse来返回JSON数据，JSONResponse是Response的子类，它的content属性接收一个dict类型。

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI()


class User(BaseModel):
    username: str
    user_code: int = Field(alias='userCode')


@app.get('/user')
async def get_user():
    user = User(username='Tom', userCode='123')
    return JSONResponse(content=user.dict(by_alias=True))

一般来说，如果不需要设置额外信息，我们直接返回dict或pydantic模型对象就可以了，使用JSONResponse通常是为了设置一些额外信息，例如特殊的响应状态码、响应头等。

设置HTTP状态码

FastAPI中我们可以通过注入的Response对象设置响应状态码。

from fastapi import FastAPI, Response

app = FastAPI()


@app.get('/foo')
async def foo(response: Response):
    response.status_code = 500
    return {'message': 'failed!'}

除了注入的Response对象，在直接返回Response、JSONResponse等对象时也都可以指定状态码，下面是一个例子。

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()


@app.get('/foo')
async def foo():
    return Response(status_code=500)

注意：这个Response和上面代码中的Response不是同一个类。

返回HTTP重定向

FastAPI服务端可以通过RedirectResponse返回重定向。

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get('/foo')
async def foo():
    return RedirectResponse(url='/')

输出文件

FastAPI服务端可以通过FileResponse返回文件，不过这里我们通常需要正确设置MIME信息后浏览器才能正确加载。

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()


@app.get('/download')
async def download():
    return FileResponse(path='C:/1.jpg', media_type='image/jpeg')

如果还需要指定Content-Disposition响应头（供浏览器识别并弹出下载对话框），我们还需要使用FileResponse的filename属性。

输出数据流

除了FileResponse，我们还可以使用StreamingResponse输出数据，它需要一个可迭代对象作为参数，我们这里使用一个异步生成器来读取图片并写入响应。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get('/download')
async def download():
    async def read_file():
        with open('C:/1.jpg', mode="rb") as fp:
            chunk = fp.read(1024)
            while chunk:
                yield chunk
                chunk = fp.read(1024)

    return StreamingResponse(read_file(), media_type='image/jpeg')

StreamingResponse还可用于输出Server-Sent Events（SSE），下面是一个例子。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get('/stream')
async def stream():
    async def output_data():
        for i in range(0, 5):
            yield f'data: {i}\n\n'

    return StreamingResponse(output_data(), media_type='text/event-stream')

设置响应头信息

响应头信息可以通过注入的Response对象获取。

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()


@app.get('/foo')
async def foo(response: Response):
    response.headers['foo'] = 'bar'
    return {'message': 'ok'}

设置Cookie

和响应头类似，设置Cookie也是通过注入的Response对象来操作的。

from fastapi import FastAPI
from fastapi.responses import Response

app = FastAPI()


@app.get('/foo')
async def foo(response: Response):
    response.set_cookie('foo', 'bar')
    return {'message': 'ok'}

除了设置Cookie，Response对象也支持删除Cookie，我们调用delete_cookie()方法即可。

作者：Gacfox