我正在使用一个自定义的JWT身份验证中间件来验证JWT。
import jwt
from urllib.parse import parse_qs
from channels.db import database_sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.conf import settings
from rest_framework_simplejwt.tokens import AccessToken
@database_sync_to_async
def get_user(user_id):
User = get_user_model()
try:
user = User.objects.get(id=user_id)
return user
except User.DoesNotExist:
return AnonymousUser()
class TokenAuthMiddleware:
"""
Custom middleware (insecure) that takes user IDs from the query string.
"""
def __init__(self, inner):
# Store the ASGI application we were passed
self.inner = inner
async def __call__(self, scope, receive, send):
# Look up user from query string (you should also do things like
# checking if it is a valid user ID, or if scope["user"] is already
# populated).
token = parse_qs(scope["query_string"].decode())["token"][0]
AccessToken(token).verify()
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256'])
scope["user"] = await get_user(int(payload["user_id"]))
return await self.inner(dict(scope), receive, send)
我收到错误TypeError: object.__init__() takes exactly one argument (the instance to initialize)
我遵循了的官方文档
有人能告诉我问题到底在哪里吗？
对于任何面临相同问题的人来说，问题实际上不在中间件中。在我的案例中，问题出现在asgi.py中。我忘了在consumer上调用.as_asgi()方法。
application = ProtocolTypeRouter(
{
"http": get_asgi_application(),
# Just HTTP for now. (We can add other protocols later.)
"websocket": TokenAuthMiddlewareStack(
URLRouter([
path("chat/", ChatConsumer.as_asgi()),
])
)
}
)
注意ChatConsumer上的as_asgi()方法。我错过了这一点，这最终让我认为问题出在我的自定义中间件中。