class GithubUser(models.Model):
user = models.OneToOneField(
get_user_model(), on_delete=models.CASCADE, related_name="github_user"
)
account_name = models.CharField(max_length=200)
account_id = models.BigIntegerField()
account_type = models.CharField(max_length=20)
avatar_url = models.CharField(max_length=200)
created_at = models.DateTimeField(auto_now_add=True)
@property
def access_token(self):
try:
return (
self.access_tokens.exclude(expires_at__lt=timezone.now())
.latest("expires_at")
.token
)
except:
return None
def get_active_access_token(self):
access_token = self.access_token
if access_token is None:
return self.get_new_tokens()[0]
return access_token
@property
def refresh_token(self):
try:
return (
self.refresh_tokens.exclude(expires_at__lt=timezone.now())
.latest("expires_at")
.token
)
except:
return None
def process_token_response(self, token_response):
now = timezone.now()
logger.info(token_response)
data = parse_qs(token_response)
access_token = GithubUserAccessToken.objects.create(
token=data["access_token"][0],
github_user=self,
expires_at=now + datetime.timedelta(seconds=int(data["expires_in"][0])),
)
refresh_token = GithubUserRefreshToken.objects.create(
token=data["refresh_token"][0],
github_user=self,
expires_at=now
+ datetime.timedelta(seconds=int(data["refresh_token_expires_in"][0])),
)
return (access_token.token, refresh_token.token)
def get_new_tokens(self):
refresh_token = self.refresh_tokens.exclude(
expires_at__lt=timezone.now()
).latest("expires_at")
response = requests.post(
"https://github.com/login/oauth/access_token",
data={
"refresh_token": refresh_token.token,
"grant_type": "refresh_token",
"client_id": settings.GITHUB_CREDS["client_id"],
"client_secret": settings.GITHUB_CREDS["client_secret"],
},
)
return self.process_token_response(response.content.decode())
class GithubUser(models.Model):
user = models.OneToOneField(
get_user_model(), on_delete=models.CASCADE, related_name="github_user"
)
account_name = models.CharField(max_length=200)
account_id = models.BigIntegerField()
account_type = models.CharField(max_length=20)
avatar_url = models.CharField(max_length=200)
created_at = models.DateTimeField(auto_now_add=True)
@property
def access_token(self):
try:
return (
self.access_tokens.exclude(expires_at__lt=timezone.now())
.latest("expires_at")
.token
)
except:
return None
def get_active_access_token(self):
access_token = self.access_token
if access_token is None:
return self.get_new_tokens()[0]
return access_token
@property
def refresh_token(self):
try:
return (
self.refresh_tokens.exclude(expires_at__lt=timezone.now())
.latest("expires_at")
.token
)
except:
return None
def process_token_response(self, token_response):
now = timezone.now()
logger.info(token_response)
data = parse_qs(token_response)
access_token = GithubUserAccessToken.objects.create(
token=data["access_token"][0],
github_user=self,
expires_at=now + datetime.timedelta(seconds=int(data["expires_in"][0])),
)
refresh_token = GithubUserRefreshToken.objects.create(
token=data["refresh_token"][0],
github_user=self,
expires_at=now
+ datetime.timedelta(seconds=int(data["refresh_token_expires_in"][0])),
)
return (access_token.token, refresh_token.token)
def get_new_tokens(self):
refresh_token = self.refresh_tokens.exclude(
expires_at__lt=timezone.now()
).latest("expires_at")
response = requests.post(
"https://github.com/login/oauth/access_token",
data={
"refresh_token": refresh_token.token,
"grant_type": "refresh_token",
"client_id": settings.GITHUB_CREDS["client_id"],
"client_secret": settings.GITHUB_CREDS["client_secret"],
},
)
return self.process_token_response(response.content.decode())