Skip to content

GitHub Services (API)

Encapsulates interactions with the GitHub GraphQL API using github_api_toolkit and a GitHub App installation token.

Overview

  • Retrieves a GitHub App installation token via AWS Secrets Manager (AWS_SECRET_NAME) and the provided GITHUB_APP_CLIENT_ID.
  • Builds a GraphQL client interface for requests.
  • Provides get_all_user_details() which returns:
  • user_to_email: username → list of verified org emails
  • email_to_user: email → username
  • user_to_id: username → GitHub account ID
  • Or a tuple ("NotFound", <message>) if the org is missing/inaccessible.

Quick Start

import boto3
from github_services import GitHubServices
from logger import wrapped_logging

logger = wrapped_logging(False)
sm = boto3.client("secretsmanager")

svc = GitHubServices(
    org="<org>",
    logger=logger,
    secret_manager=sm,
    secret_name="<aws_secret_name>",
    app_client_id="<github_app_client_id>",
)

result = svc.get_all_user_details()
if isinstance(result, tuple) and result[0] == "NotFound":
    logger.log_error(f"Org issue: {result[1]}")
else:
    user_to_email, email_to_user, user_to_id = result

Notes

  • Skips members with no verified org emails and logs a warning.
  • Paginates through org members in batches of 100.
  • Errors retrieving tokens or invalid secrets raise exceptions.

Reference

GitHubServices

Source code in src/github_services.py 
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class GitHubServices:
    def __init__(
        self,
        org: str,
        logger: Any,
        secret_manager: Any,
        secret_name: str,
        app_client_id: str,
    ):
        """
        Initialises the GitHub Services Class

        Raises:
            Exception: if GitHub app installation token is not found

        Args:
            org - Organisation name
            logger - The Lambda functions logger
            secret_manager - The S3 secrets manager
            secret_name - Secret name for AWS
            app_client_id - GitHub App Client ID
        """

        self.org = org
        self.logger = logger

        token = self.get_access_token(secret_manager, secret_name, app_client_id)

        # Ensure we have a valid token tuple before proceeding
        if not isinstance(token, tuple):
            self.logger.log_error(
                f"Failed to retrieve GitHub App installation token: {token}"
            )
            raise Exception(str(token))

        access_token = token[0]

        self.ql = github_api_toolkit.github_graphql_interface(access_token)

    def get_access_token(
        self, secret_manager: Any, secret_name: str, app_client_id: str
    ) -> Tuple[str, str]:
        """Gets the access token from the AWS Secret Manager.

        Args:
            secret_manager (Any): The Boto3 Secret Manager client.
            secret_name (str): The name of the secret to get.
            app_client_id (str): The client ID of the GitHub App.

        Raises:
            Exception: If the secret is not found in the Secret Manager.
            Exception: if GitHub app installation token is not found

        Returns:
            str: GitHub token.
        """
        response = secret_manager.get_secret_value(SecretId=secret_name)

        pem_contents = response.get("SecretString", "")

        if not pem_contents:
            error_message = f"Secret {secret_name} not found in AWS Secret Manager. Please check your environment variables."
            self.logger.log_error(error_message)
            raise Exception(error_message)

        token = github_api_toolkit.get_token_as_installation(
            self.org, pem_contents, app_client_id
        )

        if not isinstance(token, tuple):
            self.logger.log_error(
                f"Failed to retrieve GitHub App installation token: {token}"
            )
            raise Exception(str(token))

        return token

    def get_all_user_details(self) -> tuple[dict, dict, dict] | tuple:
        """
        Retrieve all the usernames within the GitHub organisation

        Returns:
            list(dict) - members usernames, emails and account ids
        """

        user_to_email = {}
        email_to_user = {}
        user_to_id = {}
        has_next_page = True
        cursor = None

        while has_next_page:
            query = """
                query ($org: String!, $cursor: String) {
                    organization(login: $org) {
                        membersWithRole(first: 100, after: $cursor) {
                            pageInfo {
                                hasNextPage
                                endCursor
                            }
                            nodes {
                                login
                                databaseId
                                organizationVerifiedDomainEmails(login: $org)
                            }
                        }
                    }
                }
            """

            params = {"org": self.org, "cursor": cursor}

            # Use instance-aware request (passes headers/token and has fallback)
            response_json = self.ql.make_ql_request(query, params).json()

            org_data = response_json.get("data", {}).get("organization")

            if not org_data:
                org_error_message = (
                    f"Organisation '{self.org} not found or inaccessible'"
                )
                self.logger.log_error(org_error_message)
                return ("NotFound", org_error_message)

            members_conn = org_data.get("membersWithRole", {})
            page_info = members_conn.get("pageInfo", {})
            has_next_page = page_info.get("hasNextPage", False)
            cursor = page_info.get("endCursor")

            for node in members_conn.get("nodes", []):
                username = node.get("login")
                account_id = node.get("databaseId")
                emails = node.get("organizationVerifiedDomainEmails", [])

                if not username:
                    self.logger.log_warning("Skipping member with empty username")
                    continue

                if emails == [] or not emails:
                    self.logger.log_warning(
                        f"Skipping member '{username}' with no verified domain emails"
                    )
                    continue

                user_to_email[username] = emails
                if account_id is not None:
                    user_to_id[username] = account_id
                for address in emails:
                    email_to_user[address] = username

        return user_to_email, email_to_user, user_to_id

__init__(org, logger, secret_manager, secret_name, app_client_id)

Initialises the GitHub Services Class

Raises:

Type Description
Exception

if GitHub app installation token is not found
Source code in src/github_services.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    org: str,
    logger: Any,
    secret_manager: Any,
    secret_name: str,
    app_client_id: str,
):
    """
    Initialises the GitHub Services Class

    Raises:
        Exception: if GitHub app installation token is not found

    Args:
        org - Organisation name
        logger - The Lambda functions logger
        secret_manager - The S3 secrets manager
        secret_name - Secret name for AWS
        app_client_id - GitHub App Client ID
    """

    self.org = org
    self.logger = logger

    token = self.get_access_token(secret_manager, secret_name, app_client_id)

    # Ensure we have a valid token tuple before proceeding
    if not isinstance(token, tuple):
        self.logger.log_error(
            f"Failed to retrieve GitHub App installation token: {token}"
        )
        raise Exception(str(token))

    access_token = token[0]

    self.ql = github_api_toolkit.github_graphql_interface(access_token)

get_access_token(secret_manager, secret_name, app_client_id)

Gets the access token from the AWS Secret Manager.

Parameters:

Name Type Description Default
secret_manager Any

The Boto3 Secret Manager client.

 required
secret_name str

The name of the secret to get.

 required
app_client_id str

The client ID of the GitHub App.

 required

Raises:

Type Description
Exception

If the secret is not found in the Secret Manager.
Exception

if GitHub app installation token is not found

Returns:

Name Type Description
str Tuple[str, str]

GitHub token.
Source code in src/github_services.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get_access_token(
    self, secret_manager: Any, secret_name: str, app_client_id: str
) -> Tuple[str, str]:
    """Gets the access token from the AWS Secret Manager.

    Args:
        secret_manager (Any): The Boto3 Secret Manager client.
        secret_name (str): The name of the secret to get.
        app_client_id (str): The client ID of the GitHub App.

    Raises:
        Exception: If the secret is not found in the Secret Manager.
        Exception: if GitHub app installation token is not found

    Returns:
        str: GitHub token.
    """
    response = secret_manager.get_secret_value(SecretId=secret_name)

    pem_contents = response.get("SecretString", "")

    if not pem_contents:
        error_message = f"Secret {secret_name} not found in AWS Secret Manager. Please check your environment variables."
        self.logger.log_error(error_message)
        raise Exception(error_message)

    token = github_api_toolkit.get_token_as_installation(
        self.org, pem_contents, app_client_id
    )

    if not isinstance(token, tuple):
        self.logger.log_error(
            f"Failed to retrieve GitHub App installation token: {token}"
        )
        raise Exception(str(token))

    return token

get_all_user_details()

Retrieve all the usernames within the GitHub organisation

Returns:

Type Description
tuple[dict, dict, dict] | tuple

list(dict) - members usernames, emails and account ids
Source code in src/github_services.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def get_all_user_details(self) -> tuple[dict, dict, dict] | tuple:
    """
    Retrieve all the usernames within the GitHub organisation

    Returns:
        list(dict) - members usernames, emails and account ids
    """

    user_to_email = {}
    email_to_user = {}
    user_to_id = {}
    has_next_page = True
    cursor = None

    while has_next_page:
        query = """
            query ($org: String!, $cursor: String) {
                organization(login: $org) {
                    membersWithRole(first: 100, after: $cursor) {
                        pageInfo {
                            hasNextPage
                            endCursor
                        }
                        nodes {
                            login
                            databaseId
                            organizationVerifiedDomainEmails(login: $org)
                        }
                    }
                }
            }
        """

        params = {"org": self.org, "cursor": cursor}

        # Use instance-aware request (passes headers/token and has fallback)
        response_json = self.ql.make_ql_request(query, params).json()

        org_data = response_json.get("data", {}).get("organization")

        if not org_data:
            org_error_message = (
                f"Organisation '{self.org} not found or inaccessible'"
            )
            self.logger.log_error(org_error_message)
            return ("NotFound", org_error_message)

        members_conn = org_data.get("membersWithRole", {})
        page_info = members_conn.get("pageInfo", {})
        has_next_page = page_info.get("hasNextPage", False)
        cursor = page_info.get("endCursor")

        for node in members_conn.get("nodes", []):
            username = node.get("login")
            account_id = node.get("databaseId")
            emails = node.get("organizationVerifiedDomainEmails", [])

            if not username:
                self.logger.log_warning("Skipping member with empty username")
                continue

            if emails == [] or not emails:
                self.logger.log_warning(
                    f"Skipping member '{username}' with no verified domain emails"
                )
                continue

            user_to_email[username] = emails
            if account_id is not None:
                user_to_id[username] = account_id
            for address in emails:
                email_to_user[address] = username

    return user_to_email, email_to_user, user_to_id