FastAPI中的Request和Response是用于处理HTTP请求和响应的两个核心对象。它们提供了许多功能和属性,使你能够更灵活地处理和构建HTTP请求和响应
你可以在视图函数中声明一个变量,并指定类型为Request,那么你就可以操作这个Request对象了。如下:
from fastapi import Request
@app.get("/home")
async def home(request: Request):
print(type(request), request)
return {'code': 1}
Request具有如下属性:
app: 程序入口对象,即FastAPI对象url: 获取当前完整的url,默认情况下返回一个starlette.datastructures.URL对象。例如: http://localhost:8000/home?name=laozhangbase_url: 获取请求的路径,默认情况下返回一个starlette.datastructures.URL对象。例如:http://localhost:8000/headers: 获取请求的标头,默认情况下返回一个starlette.datastructures.Headers对象query_params: 获取查询参数,默认情况下返回一个starlette.datastructures.QueryParams对象path_params: 路径参数,默认情况下返回一个dict类型cookies: cookie信息,默认情况下返回一个dict类型client: 客户端信息,默认情况下返回一个starlette.datastructures.Address类型。例如:Address(host='127.0.0.1', port=63902)session: session信息,默认情况下返回一个dict类型。值得注意的是,在FastAPI中,使用session需要使用到starlette-session库,并且需要将SessionMiddleware添加到中间件中,否则将会引发报错,使用如下:from redis import Redis
from fastapi import FastAPI
from fastapi import Request
from starlette_session import SessionMiddleware
from starlette_session.backends import BackendType
app = FastAPI()
r = Redis.from_url("redis://:12345678@localhost:6379/1")
app.add_middleware(
SessionMiddleware,
secret_key="secret",
cookie_name="cookie22",
backend_type=BackendType.redis,
backend_client=r
)
@app.get("/home")
async def home(request: Request):
print(type(request.session), request.session)
return {'code': 1}
auth: 获取认证信息,默认情况下返回一个starlette.authentication.AuthCredentials类型。使用该属性之前,需要将AuthenticationMiddleware添加到中间件当中,否则将会引发报错,使用如下:from starlette.authentication import AuthCredentials
from starlette.authentication import SimpleUser
from starlette.authentication import UnauthenticatedUser
from starlette.authentication import AuthenticationError
from starlette.authentication import AuthenticationBackend
from starlette.middleware.authentication import AuthenticationMiddleware
class CustomAuthBackend(AuthenticationBackend):
async def authenticate(self, conn):
if 'Authorization' not in conn.headers:
return
auth = conn.headers['Authorization']
try:
scheme, credentials = auth.split()
if scheme.lower() != 'Bearer':
return AuthCredentials(), UnauthenticatedUser()
username = self.decode_credentials(credentials)
except Exception:
raise AuthenticationError('Invalid bearer auth credentials')
return AuthCredentials(['authenticated']), SimpleUser(username)
def decode_credentials(self, content):
"""
解析并获取用户信息
"""
return 'laozhang'
app.add_middleware(AuthenticationMiddleware, backend=CustomAuthBackend())
@app.get("/home")
async def home(request: Request):
print(type(request.auth), request.auth)
return {'code': 1}
user: 获取认证的用户对象,默认情况下返回一个starlette.authentication.BaseUser对象。使用该属性之前,需要将AuthenticationMiddleware添加到中间件当中,否则将会引发报错。state: 获取状态对象,默认情况下返回一个starlette.datastructures.State对象method: 获取请求方法,默认情况下返回一个str对象,例如:GET/POST等receive: 获取响应接收方法Request具有如下方法:
url_for(): 获取指定名称的URL,默认情况下返回一个starlette.datastructures.URL对象@app.get("/home")
async def home(request: Request):
res = request.url_for('home')
print(type(res), res) # http://localhost:8000/home
return {'code': 1}
stream(): 获取请求流,默认情况下返回一个AsyncGenerator对象body(): 获取请求体,默认情况下返回一个bytes对象json(): 获取json参数,默认情况下返回一个dict对象。值得注意的是,若没有获取到json参数,将会引发json.decoder.JSONDecodeError错误(fastapi==0.100.0)form(): 获取form表单参数,默认情况下返回一个starlette.datastructures.FormData对象。并且该方法还接收两个参数:max_files最多上传的文件数量,默认为1000,max_fields最多传递的字段数量,默认为1000close(): 关闭form表单文件上传流is_disconnected(): 获取是否为未连接状态,默认情况下返回一个bool对象你可以在视图函数中声明一个变量,并指定类型为Response,那么你就可以操作这个Response对象了,最后在视图函数返回该对象即可。使用如下:
from fastapi import Response
@app.get("/home")
async def home(resp: Response):
resp.status_code = 400
resp.body = b'success'
return resp
或者你也可以这样:
from fastapi import Response
@app.get("/home")
async def home():
resp = Response(status_code=400, content='success')
return resp
Response接收如下参数:
content: 响应内容,接收任何类型的数据,默认为Nonestatus_code: 响应状态码,接收int类型的数据,默认为200headers: 响应头,接收一个Mapping[str, str]类型的数据,默认为Nonemedia_type: 媒体类型,接收一个str类型的数据,默认为Nonebackground: 后台任务,接收一个starlette.background.BackgroundTask对象,默认为NoneResponse具有如下属性:
media_type: 媒体类型,返回一个str类型数据charset: 编码方式,返回一个str类型数据status_code: 响应状态码,返回一个int类型数据background: 后台任务,返回一个starlette.background.BackgroundTask对象body: 响应体,返回一个bytes类型数据headers: 响应头,返回一个Mapping[str, str]类型的数据Response具有如下方法:
render(): render()将响应内容编码成指定charset编码的内容,返回一个bytes数据内容。接收一个参数content为任意类型数据!init_headers(): init_headers()初始化响应头,无返回值。接收一个Mapping[str, str]类型的数据set_cookie(): 为响应设置cookiedelete_cookie(): 删除cookieHTML内容响应
from fastapi.responses import HTMLResponse
@app.get("/home")
async def home():
resp = HTMLResponse(status_code=200, content='这是一个标题
')
return resp
文本内容响应
from fastapi.responses import PlainTextResponse
@app.get("/home")
async def home():
resp = PlainTextResponse(status_code=200, content='这是一个标题')
return resp
JSON内容响应
from fastapi.responses import JSONResponse
@app.get("/home")
async def home():
resp = JSONResponse({'code': 1, 'msg': 'success'})
return resp
ORJSONResponse是一个使用orjson的快速的可选 JSON 响应
UJSONResponse是一个使用ujson的可选 JSON 响应
返回 HTTP 重定向。默认情况下使用 307 状态代码(临时重定向)
from fastapi.responses import RedirectResponse
@app.get("/home")
async def home():
resp = RedirectResponse('https://www.baidu.com')
return resp
采用异步生成器或普通生成器/迭代器,然后流式传输响应主体
from fastapi.responses import StreamingResponse
async def body_generator():
for _ in range(10):
yield "some content"
@app.get("/home")
async def home():
resp = StreamingResponse(body_generator())
return resp
异步传输文件作为响应
from fastapi.responses import FileResponse
@app.get("/home")
async def home():
resp = FileResponse('__init__.py')
return resp
FileResponse与其他响应不同的是,它接收如下参数:
path: 文件路径,接收一个str或者PathLike对象数据status_code: 响应状态码headers: 响应头,接收一个Mapping[str, str]类型的数据,默认为Nonemedia_type: 媒体类型,接收一个str类型的数据,默认为Nonebackground: 后台任务,接收一个starlette.background.BackgroundTask对象,默认为Nonefilename: 文件名称,接收一个str类型数据stat_result: 接收一个os.stat_result对象数据method: 方法,接收一个strcontent_disposition_type: 接收一个str类型数据,默认是attachment