具有依赖注入的 FastAPI 身份验证

具有依赖注入的 fastapi 身份验证

fastapi 是一个用于在 python 中构建 api 的现代 web 框架。它是我个人最喜欢的 web 框架之一，因为它内置了对 openapi 规范的支持（这意味着您可以编写后端代码并从中生成所有内容），并且它支持依赖注入。

在这篇文章中，我们将简要介绍一下 fastapi 的 depends 是如何工作的。然后我们将了解为什么它如此适用于身份验证和授权。我们还将它与中间件进行对比，中间件是身份验证的另一个常见选项。最后，我们将了解 fastapi 中一些更高级的授权模式。

什么是依赖注入？

fastapi 更强大的功能之一是它对 依赖注入 的一流支持。我们有一个更长的指南这里，但让我们看一个如何使用它的快速示例。

假设我们正在构建一个分页 api。每个api调用可以包括page_number和page_size。现在，我们可以创建一个 api 并直接传入这些参数：

@app.get("/things/")
async def fetch_things(page_number: int = 0, page_size: int = 100):
    return db.fetch_things(page_number, page_size)

但是，我们可能想添加一些验证逻辑，这样就没有人要求 page_number -1 或 page_size 10,000,000。

@app.get("/things/")
async def fetch_things(page_number: int = 0, page_size: int = 100):
    if page_number  100:
        raise httpexception(status_code=400, detail="page size can be at most 100")
    return db.fetch_things(page_number, page_size)

这……很好，但如果我们有 10 个 api 或 100 个 api 都需要相同的分页参数，那就有点乏味了。这就是依赖注入的用武之地 - 我们可以将所有这些逻辑移至一个函数中，并将该函数注入到我们的 api 中：

async def paging_params_dep(page_number: int = 0, page_size: int = 100):
    if page_number  100:
        raise httpexception(status_code=400, detail="page size can be at most 100")
    return pagingparams(page_number, page_size)

@app.get("/things/")
async def fetch_things(paging_params: pagingparams = depends(paging_params_dep)):
    return db.fetch_things(paging_params)

@app.get("/other_things/")
async def fetch_other_things(paging_params: pagingparams = depends(paging_params_dep)):
    return db.fetch_other_things(paging_params)

这有一些不错的好处：

每条采用 pagingparams 的路由都会自动验证并具有默认值。
它比每条路由的第一行是 validate_paging_params(page_number, page_size)
这仍然适用于 fastapi 的 openapi 支持 - 这些参数将显示在您的 openapi 规范中。

这与身份验证有什么关系？

事实证明，这也是一种建模身份验证的好方法！想象一下你有一个像这样的函数：

async def validate_token(token: str):
    try:
        # this could be jwt validation, looking up a session token in the db, etc.
        return await get_user_for_token(token)
    except:
        return none

要将其连接到 api 路由，我们需要做的就是将其包装在依赖项中：

async def require_valid_token_dep(req: request):
    # this could also be a cookie, x-api-key header, etc.
    token = req.headers["authorization"]
    user = await validate_token(token)
    if user == none:
        raise httpexception(status_code=401, detail="unauthorized")
    return user

然后我们所有受保护的路由都可以添加此依赖项：

@app.get("/protected")
async def do_secret_things(user: user = depends(require_valid_token_dep)):
    # do something with the user

如果用户提供了有效的令牌，则该路由将运行并设置用户。否则，将返回 401。

注意：openapi/swagger 确实对指定身份验证令牌具有一流的支持，但您必须使用其中一个专用类。您可以使用 fastapi.security 中的 httpbearer(auto_error=false) 来代替 req.headers["authorization"]，它会返回 httpauthorizationcredentials。

中间件与 depends for auth

fastapi 与大多数框架一样，有一个中间件 的概念。您的中间件可以包含将在请求之前和之后运行的代码。它可以在请求到达您的路由之前修改请求，也可以在响应返回给用户之前修改响应。

在许多其他框架中，中间件是进行身份验证检查的非常常见的地方。然而，这通常是因为中间件还负责将用户“注入”到路由中。例如，express 中的常见模式是执行以下操作：

app.get("/protected", authmiddleware, (req, res) => {
    // req.user is set by the middleware
    // as there's no good way to pass in extra information into this route,
    // outside of the request
});

由于fastapi具有内置的注入概念，因此您可能根本不需要使用中间件。如果您需要定期“刷新”您的身份验证令牌（以使其保持活动状态）并将响应设置为 cookie，我会考虑使用中间件。

在这种情况下，您需要使用 request.state 将信息从中间件传递到路由（如果您愿意，您可以使用依赖项来验证 request.state）。

否则，我会坚持使用 depends，因为用户将被直接注入到您的路由中，而不需要通过 request.state。

授权 - 多租户、角色和权限

如果我们应用迄今为止所学到的一切，添加多租户、角色或权限可能会非常简单。假设我们为每个客户都有一个唯一的子域，我们可以为该子域建立依赖关系：

async def tenant_by_subdomain_dep(request: request) -> optional[str]:
    # first we get the subdomain from the host header
    host = request.headers.get("host", "")
    parts = host.split(".")
    if len(parts) 



<p>我们可以将这个想法与之前的想法结合起来，制作一个新的“多租户”依赖：<br></p>

async def get_user_and_tenant_for_token(
    user: user = depends(require_valid_token_dep),
    tenant: tenant = depends(tenant_by_subdomain_dep),
) -> userandtenant:
    is_user_in_tenant = await check_user_is_in_tenant(tenant, user)
    if is_user_in_tenant:
        return userandtenant(user, tenant)
    raise httpexception(status_code=403, detail="forbidden")




然后我们可以将此依赖项注入到我们的路由中：


@app.get("/protected")
async def do_secret_things(user_and_tenant: userandtenant = depends(get_user_and_tenant_for_token)):
    # do something with the user and tenant




这最终会做一些主要的事情：

检查用户是否拥有有效令牌
检查用户是否正在向有效的子域发出请求
检查用户是否应该有权访问该子域

如果不满足任何这些不变量 - 将返回错误并且我们的路线将永远不会运行。我们可以扩展它以包括其他内容，例如角色和权限 (rbac) 或确保用户具有特定的属性集（有效付费订阅与无有效订阅）。

propelauth

在 propelauth，我们是 fastapi 的忠实粉丝。我们有一个 fastapi 库，使您能够快速设置身份验证和授权 - 包括 sso、企业 sso / saml、scim 配置等。

这一切都与您在上面看到的依赖项一起工作，例如：

@app.get("/")
async def root(current_user: User = Depends(auth.require_user)):
    return {"message": f"Hello {current_user.user_id}"}

您可以在这里了解更多信息。

总结

fastapi 的依赖注入 提供了一种强大的方法来处理 web 应用程序中的身份验证和授权。
depends 功能允许使用干净、可重用的代码来验证令牌、检查用户权限和处理多租户。
与中间件相比，使用依赖项进行身份验证提供了更大的灵活性以及与路由功能的直接集成。
复杂的授权场景，例如多租户和基于角色的访问控制可以使用嵌套依赖项有效地实现。
propelauth 提供了一个 fastapi 库，可以简化高级身份验证和授权功能的实现。

以上就是具有依赖注入的 FastAPI 身份验证的详细内容，更多请关注其它相关文章！