from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
return {"token": token}
但这并不实用。
接下来,我们学习如何返回当前用户。
创建用户模型
首先,创建 Pydantic 用户模型。
与使用 Pydantic 声明请求体相同,并且可在任何位置使用:
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: Union[str, None] = None
full_name: Union[str, None] = None
disabled: Union[bool, None] = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
创建 get_current_user
依赖项
创建 get_current_user
依赖项。
还记得依赖项支持子依赖项吗?
get_current_user
使用 oauth2_scheme
作为依赖项。
与之前直接在路径操作中的做法相同,新的 get_current_user
依赖项从子依赖项 oauth2_scheme
中接收 str
类型的 token
:
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: Union[str, None] = None
full_name: Union[str, None] = None
disabled: Union[bool, None] = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
获取用户
get_current_user
使用创建的(伪)工具函数,该函数接收 str
类型的令牌,并返回 Pydantic 的 User
模型:
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: Union[str, None] = None
full_name: Union[str, None] = None
disabled: Union[bool, None] = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
注入当前用户
在路径操作 的 Depends
中使用 get_current_user
:
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: Union[str, None] = None
full_name: Union[str, None] = None
disabled: Union[bool, None] = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
注意,此处把 current_user
的类型声明为 Pydantic 的 User
模型。
这有助于在函数内部使用代码补全和类型检查。
还记得请求体也是使用 Pydantic 模型声明的吧。
放心,因为使用了 Depends
,FastAPI 不会搞混。
依赖系统的这种设计方式可以支持不同的依赖项返回同一个 User
模型。
而不是局限于只能有一个返回该类型数据的依赖项。
其它模型
接下来,直接在路径操作函数中获取当前用户,并用 Depends
在依赖注入系统中处理安全机制。
开发者可以使用任何模型或数据满足安全需求(本例中是 Pydantic 的 User
模型)。
而且,不局限于只能使用特定的数据模型、类或类型。
不想在模型中使用 username
,而是使用 id
和 email
?当然可以。这些工具也支持。
只想使用字符串?或字典?甚至是数据库类模型的实例?工作方式都一样。
实际上,就算登录应用的不是用户,而是只拥有访问令牌的机器人、程序或其它系统?工作方式也一样。
尽管使用应用所需的任何模型、类、数据库。FastAPI 通过依赖注入系统都能帮您搞定。
代码大小
这个示例看起来有些冗长。毕竟这个文件同时包含了安全、数据模型的工具函数,以及路径操作等代码。
但,关键是:
安全和依赖注入的代码只需要写一次。
就算写得再复杂,也只是在一个位置写一次就够了。所以,要多复杂就可以写多复杂。
但是,就算有数千个端点(路径操作),它们都可以使用同一个安全系统。
而且,所有端点(或它们的任何部件)都可以利用这些依赖项或任何其它依赖项。
所有路径操作只需 3 行代码就可以了:
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class User(BaseModel):
username: str
email: Union[str, None] = None
full_name: Union[str, None] = None
disabled: Union[bool, None] = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="[email protected]", full_name="John Doe"
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
现在,我们可以直接在路径操作函数中获取当前用户。
至此,安全的内容已经讲了一半。
只要再为用户或客户端的路径操作添加真正发送 username
和 password
的功能就可以了。
下一章见。
The FastAPI trademark is owned by @tiangolo and is registered in the US and across other regions
Made with
Material for MkDocs