获取当前用户¶
在上一章中，安全系统（基于依赖注入系统）向路径操作函数提供了一个 str 类型的 token。
from typing import Annotated
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: Annotated[str, Depends(oauth2_scheme)]):
return {"token": token}
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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 Annotated
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
)
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
return current_user
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", 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 Annotated
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
)
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
return current_user
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", 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 类型的 token 并返回我们的 Pydantic User 模型。
from typing import Annotated
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
)
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
return current_user
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", 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 Annotated
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
)
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
return current_user
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", 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 模型声明的。
在这里，FastAPI 不会感到困惑，因为你使用了 Depends。
提示
这种依赖注入系统的设计方式允许我们拥有返回 User 模型的不同依赖项（不同的“可依赖对象”）。
我们并不局限于只能有一个能返回该类型数据的依赖项。
其他模型¶
现在，你可以直接在路径操作函数中获取当前用户，并在依赖注入层面处理安全机制，使用 Depends 即可。
而且你可以对安全需求使用任何模型或数据（在本例中是 Pydantic 模型 User）。
你并不受限于使用特定的数据模型、类或类型。
你想拥有 id 和 email 而模型中没有 username？没问题。你可以使用同样的工具。
你只想拥有一个 str？或者仅仅是一个 dict？或者直接使用数据库类的模型实例？它们的工作方式完全一样。
你的应用程序中没有登录用户，而是只有访问令牌的机器人、爬虫或其他系统？同样，它们的工作方式也完全一样。
只需使用应用程序所需的任何类型的模型、类或数据库。FastAPI 的依赖注入系统都能满足你的需求。
代码量¶
这个例子看起来可能有点冗长。请记住，我们在同一个文件中混合了安全、数据模型、工具函数和路径操作。
但关键点在于：
安全和依赖注入的代码只需要写一次。
你可以根据需要将其做得尽可能复杂。但即便如此，它也只需编写一次，且位于同一个地方。并且它具备极高的灵活性。
你可以拥有成千上万个使用相同安全系统的端点（路径操作）。
它们中的每一个（或你选择的任何一部分）都可以利用这些可重用的依赖项或你创建的任何其他依赖项。
而这成千上万个路径操作每一个都可以精简到只需 3 行代码。
from typing import Annotated
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
)
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
return user
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
return current_user
🤓 其他版本和变体
提示
如果可能，请优先使用 Annotated 版本。
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: str | None = None
full_name: str | None = None
disabled: bool | None = None
def fake_decode_token(token):
return User(
username=token + "fakedecoded", email="john@example.com", 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。
接下来就是这些内容。