如何在Strawberry-GraphQL中实现Django用户身份验证
实现兼容异步Django视图的登录验证
问题背景
需要让GraphQL的同步和异步视图都强制用户登录后才能访问,原生login_required装饰器对同步视图有效,但不支持异步逻辑。自定义继承视图的方式虽可行,但代码冗余不够简洁,希望有更优雅的实现方案。
更简洁的实现方案
方案1:改造LoginRequiredMixin适配异步类视图
基于Django原生的AccessMixin封装异步兼容的登录验证Mixin,可直接继承复用:
from asgiref.sync import sync_to_async from django.contrib.auth.mixins import AccessMixin from django.contrib.auth.views import redirect_to_login from django.shortcuts import resolve_url from urllib.parse import urlparse class AsyncLoginRequiredMixin(AccessMixin): async def dispatch(self, request, *args, **kwargs): # 异步检查用户认证状态 is_authenticated = await sync_to_async(lambda: request.user.is_authenticated)() if not is_authenticated: path = request.build_absolute_uri() resolved_login_url = resolve_url(self.get_login_url()) # 对齐原生login_required的next参数处理逻辑 login_scheme, login_netloc = urlparse(resolved_login_url)[:2] current_scheme, current_netloc = urlparse(path)[:2] if (not login_scheme or login_scheme == current_scheme) and ( not login_netloc or login_netloc == current_netloc ): path = request.get_full_path() return redirect_to_login(path, resolved_login_url, self.get_redirect_field_name()) return await super().dispatch(request, *args, **kwargs)
在异步GraphQL视图中继承该Mixin:
from strawberry.django.views import AsyncGraphQLView from django.utils.decorators import method_decorator from django.views.decorators.csrf import csrf_exempt class ProtectedAsyncGraphQLView(AsyncLoginRequiredMixin, AsyncGraphQLView): @method_decorator(csrf_exempt) async def dispatch(self, request, *args, **kwargs): return await super().dispatch(request, *args, **kwargs) # urls.py配置 urlpatterns = [ path("graphql", ProtectedAsyncGraphQLView.as_view(schema=schema)), path("graphql/sync", login_required(GraphQLView.as_view(schema=schema))), ]
方案2:自定义异步版login_required装饰器
编写支持异步视图的装饰器,用法和原生login_required完全一致,灵活性更高:
from asgiref.sync import sync_to_async from django.contrib.auth.views import redirect_to_login from django.shortcuts import resolve_url from urllib.parse import urlparse from django.conf import settings def async_login_required(view_func): async def wrapped_view(request, *args, **kwargs): is_authenticated = await sync_to_async(lambda: request.user.is_authenticated)() if not is_authenticated: path = request.build_absolute_uri() resolved_login_url = resolve_url(settings.LOGIN_URL) login_scheme, login_netloc = urlparse(resolved_login_url)[:2] current_scheme, current_netloc = urlparse(path)[:2] if (not login_scheme or login_scheme == current_scheme) and ( not login_netloc or login_netloc == current_netloc ): path = request.get_full_path() return redirect_to_login(path, resolved_login_url) return await view_func(request, *args, **kwargs) return wrapped_view
直接在URL配置中使用:
from django.urls import path from strawberry.django.views import AsyncGraphQLView, GraphQLView urlpatterns = [ path("graphql/sync", login_required(GraphQLView.as_view(schema=schema))), path("graphql", async_login_required(AsyncGraphQLView.as_view(schema=schema))), ]
方案3:利用Strawberry内置权限系统(GraphQL专属)
针对Strawberry GraphQL场景,可直接用其权限系统实现细粒度控制,无需修改视图:
import strawberry from strawberry.permission import BasePermission from asgiref.sync import sync_to_async class IsAuthenticated(BasePermission): message = "用户未登录,无法访问该资源。" async def has_permission(self, source: object, info: strawberry.types.Info, **kwargs) -> bool: return await sync_to_async(lambda: info.context.request.user.is_authenticated)() # 在Schema字段/操作上绑定权限 @strawberry.type class Query: @strawberry.field(permission_classes=[IsAuthenticated]) async def protected_data(self, info: strawberry.types.Info) -> str: return f"欢迎登录,{info.context.request.user.username}!" schema = strawberry.Schema(query=Query)
该方案可精准控制单个GraphQL字段或操作的访问权限,而非全局拦截视图。
内容的提问来源于stack exchange,提问作者fabien-michel
相关产品推荐
相关产品推荐

