def check_identity(self, token):
"""
Lookup token on identity service and create/update local user.
"""
logger.info("checking identity server {}".format(settings.KEL["IDENTITY_URL"]))
params = {"access_token": token}
resp = requests.get("{}/tokeninfo/".format(settings.KEL["IDENTITY_URL"]), params=params)
if not resp.ok:
return None
payload = resp.json()
with transaction.atomic():
user = next(iter(User.objects.filter(username=payload["user"]["username"])), None)
if user is None:
user = User.objects.create(username=payload["user"]["username"])
else:
user.last_login = timezone.now()
user.save()
return user
评论列表
文章目录