|
| 1 | +import requests |
| 2 | +from django.conf import settings |
| 3 | +from requests import RequestException |
| 4 | + |
| 5 | +from custom_auth.oauth.exceptions import GithubError |
| 6 | +from custom_auth.oauth.helpers.github_helpers import convert_response_data_to_dictionary, get_first_and_last_name |
| 7 | +from util.logging import get_logger |
| 8 | + |
| 9 | +GITHUB_API_URL = "https://api.github.com" |
| 10 | +GITHUB_OAUTH_URL = "https://github.com/login/oauth" |
| 11 | + |
| 12 | +NON_200_ERROR_MESSAGE = "Github returned {} status code when getting an access token." |
| 13 | + |
| 14 | +logger = get_logger(__name__) |
| 15 | + |
| 16 | + |
| 17 | +class GithubUser: |
| 18 | + def __init__(self, code: str, client_id: str = None, client_secret: str = None): |
| 19 | + self.client_id = client_id or settings.GITHUB_CLIENT_ID |
| 20 | + self.client_secret = client_secret or settings.GITHUB_CLIENT_SECRET |
| 21 | + |
| 22 | + self.access_token = self._get_access_token(code) |
| 23 | + self.headers = { |
| 24 | + "Authorization": f"token {self.access_token}" |
| 25 | + } |
| 26 | + |
| 27 | + def _get_access_token(self, code) -> str: |
| 28 | + data = { |
| 29 | + "code": code, |
| 30 | + "client_id": self.client_id, |
| 31 | + "client_secret": self.client_secret |
| 32 | + } |
| 33 | + response = requests.post(f"{GITHUB_OAUTH_URL}/access_token", data=data) |
| 34 | + |
| 35 | + if response.status_code != 200: |
| 36 | + raise GithubError(NON_200_ERROR_MESSAGE.format(response.status_code)) |
| 37 | + |
| 38 | + response_json = convert_response_data_to_dictionary(response.text) |
| 39 | + if "error" in response_json: |
| 40 | + error_message = response_json["error_description"].replace("+", " ") |
| 41 | + raise GithubError(error_message) |
| 42 | + |
| 43 | + return response_json["access_token"] |
| 44 | + |
| 45 | + def get_user_info(self) -> dict: |
| 46 | + # TODO: use threads? |
| 47 | + try: |
| 48 | + return { |
| 49 | + **self._get_user_name_and_id(), |
| 50 | + "email": self._get_primary_email() |
| 51 | + } |
| 52 | + except RequestException: |
| 53 | + raise GithubError("Failed to communicate with the Github API.") |
| 54 | + |
| 55 | + def _get_user_name_and_id(self): |
| 56 | + user_response = requests.get(f"{GITHUB_API_URL}/user", headers=self.headers) |
| 57 | + user_response_json = user_response.json() |
| 58 | + full_name = user_response_json.get("name") |
| 59 | + first_name, last_name = get_first_and_last_name(full_name) if full_name else ["", ""] |
| 60 | + return { |
| 61 | + "first_name": first_name, |
| 62 | + "last_name": last_name, |
| 63 | + "github_user_id": user_response_json.get("id") |
| 64 | + } |
| 65 | + |
| 66 | + def _get_primary_email(self): |
| 67 | + emails_response = requests.get(f"{GITHUB_API_URL}/user/emails", headers=self.headers) |
| 68 | + |
| 69 | + # response from github should be a list of dictionaries, this will find the first entry that is both verified |
| 70 | + # and marked as primary (there should only be one). |
| 71 | + primary_email_data = next( |
| 72 | + filter(lambda email_data: email_data["primary"] and email_data["verified"], emails_response.json()), None |
| 73 | + ) |
| 74 | + |
| 75 | + if not primary_email_data: |
| 76 | + raise GithubError("User does not have a verified email address with Github.") |
| 77 | + |
| 78 | + return primary_email_data["email"] |
0 commit comments