django-graphql-jwt
django-graphql-jwt copied to clipboard
Get Authenticated User in Model
I have a model mixin to add created_by and last_modified_by fields with the currently authenticated user. I am trying to set these fields in the save method of the mixin to help keep the code DRY. Is there currently a method to get the currently authenticated user from a model method?
I tried using a middleware and signals which works fine for the session in which the user first logs in, but neither the token_issued or token_refreshed seem to be sent on each graphql call that contains a valid token. Is there another signal that would work in this case? Should I submit a PR with a new signal that is sent on each authentication?
I also tried a Graphene middleware to extract the user from info.context.user. In this case, the info.context.user is not changed from Anonymous to the authenticated user before the first call to to save the model. Seems that a second resolve needs to be preformed before the user is injected into the context.
Is there another method I should be using or just save the created_by and last_modified_by in each mutation?
Here is the relevant code:
ModelMixin:
class TrackerMixin(models.Model):
class Meta:
abstract = True
created_at = models.DateTimeField(default=now, db_index=True)
created_by = models.ForeignKey(User,
related_name="%(class)s_created",
on_delete=models.SET_NULL,
null=True, blank=True)
last_updated_at = models.DateTimeField(default=now, db_index=True)
last_updated_by = models.ForeignKey(User,
related_name="%(class)s_last_updated",
on_delete=models.SET_NULL,
null=True, blank=True)
def save(self, *args, **kwargs):
# Handle old data without created_at value
if not self.created_at:
self.created_at = now()
self.last_updated_at = now()
current_user = get_current_user()
if current_user and current_user.pk:
if not self.pk:
self.created_by = current_user
self.last_updated_by = current_user
super().save(*args, **kwargs)
Middleware:
from django.contrib.auth.models import AnonymousUser
from django.dispatch import receiver
from threading import local
from graphql_jwt.signals import token_issued, token_refreshed
_user = local()
def get_current_user():
if hasattr(_user, 'value') and _user.value:
return _user.value
@receiver([token_issued, token_refreshed])
def update_current_user_from_jwt(sender, **kwargs):
print("new sender", sender, kwargs)
class CurrentUserMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
_user.value = request.user
response = self.get_response(request)
return response
@staticmethod
def get_current_user():
if hasattr(_user, 'value') and _user.value:
return _user.value
class CurrentUserGrapheneMiddleware:
def __init__(self):
_user.value = None
def resolve(self, next, root, info, **args):
user = info.context.user
if user == AnonymousUser:
_user.value = None
else:
_user.value = user
return next(root, info, **args)
Thanks!
I was able to get the CurrentGrapheneMiddleware to update the global _user value with the authenticated user. The issue was the order that I added this custom middleware into the graphene settings. Apparently, the middlewares are run in reverse order, so I had to place the custom middleware at the beginning in order to get the user added by the graphql_jwt middleware.
Are there any concerns with this method or is there a better manner in which this capability should be implemented?
GRAPHENE = {
'SCHEMA': 'project.schema.schema',
'MIDDLEWARE': [
'project.app.middleware.CurrentUserGrapheneMiddleware',
'graphql_jwt.middleware.JSONWebTokenMiddleware',
],
}