Github

GitHubRepositoryCollaboratorsReader

Bases: BaseReader

GitHub repository collaborators reader.

Retrieves the list of collaborators of a GitHub repository and returns a list of documents.

Examples:

>>> reader = GitHubRepositoryCollaboratorsReader("owner", "repo")
>>> colabs = reader.load_data()
>>> print(colabs)

Source code in llama_index/readers/github/collaborators/base.py

class GitHubRepositoryCollaboratorsReader(BaseReader):
    """
    GitHub repository collaborators reader.

    Retrieves the list of collaborators of a GitHub repository and returns a list of documents.

    Examples:
        >>> reader = GitHubRepositoryCollaboratorsReader("owner", "repo")
        >>> colabs = reader.load_data()
        >>> print(colabs)

    """

    class FilterType(enum.Enum):
        """
        Filter type.

        Used to determine whether the filter is inclusive or exclusive.
        """

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGitHubCollaboratorsClient,
        owner: str,
        repo: str,
        verbose: bool = False,
    ):
        """
        Initialize params.

        Args:
            - github_client (BaseGitHubCollaboratorsClient): GitHub client.
            - owner (str): Owner of the repository.
            - repo (str): Name of the repository.
            - verbose (bool): Whether to print verbose messages.

        Raises:
            - `ValueError`: If the github_token is not provided and
                the GITHUB_TOKEN environment variable is not set.

        """
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._verbose = verbose

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

    def load_data(
        self,
    ) -> List[Document]:
        """
        GitHub repository collaborators reader.

        Retrieves the list of collaborators in a GitHub repository and converts them to documents.

        Each collaborator is converted to a document by doing the following:

            - The text of the document is the login.
            - The title of the document is also the login.
            - The extra_info of the document is a dictionary with the following keys:
                - login: str, the login of the user
                - type: str, the type of user e.g. "User"
                - site_admin: bool, whether the user has admin permissions
                - role_name: str, e.g. "admin"
                - name: str, the name of the user, if available
                - email: str, the email of the user, if available
                - permissions: str, the permissions of the user, if available


        :return: list of documents
        """
        documents = []
        page = 1
        # Loop until there are no more collaborators
        while True:
            collaborators: Dict = self._loop.run_until_complete(
                self._github_client.get_collaborators(
                    self._owner, self._repo, page=page
                )
            )

            if len(collaborators) == 0:
                print_if_verbose(self._verbose, "No more collaborators found, stopping")

                break
            print_if_verbose(
                self._verbose,
                f"Found {len(collaborators)} collaborators in the repo page {page}",
            )
            page += 1
            for collab in collaborators:
                extra_info = {
                    "login": collab["login"],
                    "type": collab["type"],
                    "site_admin": collab["site_admin"],
                    "role_name": collab["role_name"],
                }
                if collab.get("name") is not None:
                    extra_info["name"] = collab["name"]
                if collab.get("email") is not None:
                    extra_info["email"] = collab["email"]
                if collab.get("permissions") is not None:
                    extra_info["permissions"] = collab["permissions"]
                document = Document(
                    doc_id=str(collab["login"]),
                    text=str(collab["login"]),  # unsure for this
                    extra_info=extra_info,
                )
                documents.append(document)

            print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")

        return documents

FilterType

Bases: Enum

Filter type.

Used to determine whether the filter is inclusive or exclusive.

Source code in llama_index/readers/github/collaborators/base.py

class FilterType(enum.Enum):
    """
    Filter type.

    Used to determine whether the filter is inclusive or exclusive.
    """

    EXCLUDE = enum.auto()
    INCLUDE = enum.auto()

load_data

load_data() -> List[Document]

GitHub repository collaborators reader.

Retrieves the list of collaborators in a GitHub repository and converts them to documents.

Each collaborator is converted to a document by doing the following:

- The text of the document is the login.
- The title of the document is also the login.
- The extra_info of the document is a dictionary with the following keys:
    - login: str, the login of the user
    - type: str, the type of user e.g. "User"
    - site_admin: bool, whether the user has admin permissions
    - role_name: str, e.g. "admin"
    - name: str, the name of the user, if available
    - email: str, the email of the user, if available
    - permissions: str, the permissions of the user, if available

:return: list of documents

Source code in llama_index/readers/github/collaborators/base.py

def load_data(
    self,
) -> List[Document]:
    """
    GitHub repository collaborators reader.

    Retrieves the list of collaborators in a GitHub repository and converts them to documents.

    Each collaborator is converted to a document by doing the following:

        - The text of the document is the login.
        - The title of the document is also the login.
        - The extra_info of the document is a dictionary with the following keys:
            - login: str, the login of the user
            - type: str, the type of user e.g. "User"
            - site_admin: bool, whether the user has admin permissions
            - role_name: str, e.g. "admin"
            - name: str, the name of the user, if available
            - email: str, the email of the user, if available
            - permissions: str, the permissions of the user, if available


    :return: list of documents
    """
    documents = []
    page = 1
    # Loop until there are no more collaborators
    while True:
        collaborators: Dict = self._loop.run_until_complete(
            self._github_client.get_collaborators(
                self._owner, self._repo, page=page
            )
        )

        if len(collaborators) == 0:
            print_if_verbose(self._verbose, "No more collaborators found, stopping")

            break
        print_if_verbose(
            self._verbose,
            f"Found {len(collaborators)} collaborators in the repo page {page}",
        )
        page += 1
        for collab in collaborators:
            extra_info = {
                "login": collab["login"],
                "type": collab["type"],
                "site_admin": collab["site_admin"],
                "role_name": collab["role_name"],
            }
            if collab.get("name") is not None:
                extra_info["name"] = collab["name"]
            if collab.get("email") is not None:
                extra_info["email"] = collab["email"]
            if collab.get("permissions") is not None:
                extra_info["permissions"] = collab["permissions"]
            document = Document(
                doc_id=str(collab["login"]),
                text=str(collab["login"]),  # unsure for this
                extra_info=extra_info,
            )
            documents.append(document)

        print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")

    return documents

GitHubCollaboratorsClient

An asynchronous client for interacting with the GitHub API for collaborators.

The client supports two authentication methods: 1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var 2. GitHub App - passed as github_app_auth parameter

Examples:

>>> # Using Personal Access Token
>>> client = GitHubCollaboratorsClient("my_github_token")
>>> collaborators = client.get_collaborators("owner", "repo")
>>>
>>> # Using GitHub App
>>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
>>> app_auth = GitHubAppAuth(app_id="123", private_key=key, installation_id="456")
>>> client = GitHubCollaboratorsClient(github_app_auth=app_auth)

Source code in llama_index/readers/github/collaborators/github_client.py

class GitHubCollaboratorsClient:
    """
    An asynchronous client for interacting with the GitHub API for collaborators.

    The client supports two authentication methods:
    1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var
    2. GitHub App - passed as github_app_auth parameter

    Examples:
        >>> # Using Personal Access Token
        >>> client = GitHubCollaboratorsClient("my_github_token")
        >>> collaborators = client.get_collaborators("owner", "repo")
        >>>
        >>> # Using GitHub App
        >>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
        >>> app_auth = GitHubAppAuth(app_id="123", private_key=key, installation_id="456")
        >>> client = GitHubCollaboratorsClient(github_app_auth=app_auth)

    """

    DEFAULT_BASE_URL = "https://api.github.com"
    DEFAULT_API_VERSION = "2022-11-28"

    def __init__(
        self,
        github_token: Optional[str] = None,
        github_app_auth: Optional[Union["GitHubAppAuth", Any]] = None,
        base_url: str = DEFAULT_BASE_URL,
        api_version: str = DEFAULT_API_VERSION,
        verbose: bool = False,
    ) -> None:
        """
        Initialize the GitHubCollaboratorsClient.

        Args:
            - github_token (str, optional): GitHub token for authentication.
                If not provided, the client will try to get it from
                the GITHUB_TOKEN environment variable. Mutually exclusive with github_app_auth.
            - github_app_auth (GitHubAppAuth, optional): GitHub App authentication handler.
                Mutually exclusive with github_token.
            - base_url (str): Base URL for the GitHub API
                (defaults to "https://api.github.com").
            - api_version (str): GitHub API version (defaults to "2022-11-28").
            - verbose (bool): Whether to print verbose output (defaults to False).

        Raises:
            ValueError: If neither github_token nor github_app_auth is provided,
                       or if both are provided.

        """
        # Validate authentication parameters
        if github_token is not None and github_app_auth is not None:
            raise ValueError(
                "Cannot provide both github_token and github_app_auth. "
                "Please use only one authentication method."
            )

        self._base_url = base_url
        self._api_version = api_version
        self._verbose = verbose
        self._github_app_auth = github_app_auth
        self._github_token = None

        # Set up authentication
        if github_app_auth is not None:
            self._use_github_app = True
        else:
            self._use_github_app = False
            if github_token is None:
                github_token = os.getenv("GITHUB_TOKEN")
                if github_token is None:
                    raise ValueError(
                        "Please provide a GitHub token or GitHub App authentication. "
                        + "You can pass github_token as an argument, "
                        + "set the GITHUB_TOKEN environment variable, "
                        + "or pass github_app_auth for GitHub App authentication."
                    )
            self._github_token = github_token

        self._endpoints = {
            "getCollaborators": "/repos/{owner}/{repo}/collaborators",
        }

        # Base headers (Authorization header will be added per-request)
        self._base_headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": f"{self._api_version}",
        }

        # For backward compatibility, keep _headers with PAT token
        if not self._use_github_app:
            self._headers = {
                **self._base_headers,
                "Authorization": f"Bearer {self._github_token}",
            }
        else:
            self._headers = self._base_headers.copy()

    def get_all_endpoints(self) -> Dict[str, str]:
        """Get all available endpoints."""
        return {**self._endpoints}

    async def _get_auth_headers(self) -> Dict[str, str]:
        """Get authentication headers."""
        if self._use_github_app:
            token = await self._github_app_auth.get_installation_token()
            return {
                **self._base_headers,
                "Authorization": f"Bearer {token}",
            }
        else:
            return self._headers

    async def request(
        self,
        endpoint: str,
        method: str,
        headers: Dict[str, Any] = {},
        params: Dict[str, Any] = {},
        **kwargs: Any,
    ) -> Any:
        """
        Makes an API request to the GitHub API.

        Args:
            - `endpoint (str)`: Name of the endpoint to make the request to.
            - `method (str)`: HTTP method to use for the request.
            - `headers (dict)`: HTTP headers to include in the request.
            - `**kwargs`: Keyword arguments to pass to the endpoint URL.

        Returns:
            - `response (httpx.Response)`: Response from the API request.

        Raises:
            - ImportError: If the `httpx` library is not installed.
            - httpx.HTTPError: If the API request fails.

        Examples:
            >>> response = client.request("getCollaborators", "GET",
                                owner="owner", repo="repo", state="all")

        """
        try:
            import httpx
        except ImportError:
            raise ImportError(
                "`https` package not found, please run `pip install httpx`"
            )

        # Get authentication headers (may fetch fresh token for GitHub App)
        auth_headers = await self._get_auth_headers()
        _headers = {**auth_headers, **headers}

        _client: httpx.AsyncClient
        async with httpx.AsyncClient(
            headers=_headers, base_url=self._base_url, params=params
        ) as _client:
            try:
                response = await _client.request(
                    method, url=self._endpoints[endpoint].format(**kwargs)
                )
                response.raise_for_status()
            except httpx.HTTPError as excp:
                print(f"HTTP Exception for {excp.request.url} - {excp}")
                raise excp  # noqa: TRY201
            return response

    async def get_collaborators(
        self,
        owner: str,
        repo: str,
        page: int = 1,
    ) -> Dict:
        """
        List collaborators in a repository.

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.

        Returns:
            - See https://docs.github.com/en/rest/collaborators/collaborators?apiVersion=2022-11-28#list-repository-collaborators

        Examples:
            >>> repo_collaborators = client.get_collaborators("owner", "repo")

        """
        return (
            await self.request(
                endpoint="getCollaborators",
                method="GET",
                params={
                    "per_page": 100,
                    "page": page,
                },
                owner=owner,
                repo=repo,
            )
        ).json()

get_all_endpoints

get_all_endpoints() -> Dict[str, str]

Get all available endpoints.

Source code in llama_index/readers/github/collaborators/github_client.py

def get_all_endpoints(self) -> Dict[str, str]:
    """Get all available endpoints."""
    return {**self._endpoints}

request async

request(endpoint: str, method: str, headers: Dict[str, Any] = {}, params: Dict[str, Any] = {}, **kwargs: Any) -> Any

Makes an API request to the GitHub API.

Parameters:

Name	Type	Description	Default
- `endpoint (str)`		Name of the endpoint to make the request to.	required
- `method (str)`		HTTP method to use for the request.	required
- `headers (dict)`		HTTP headers to include in the request.	required
- `**kwargs`		Keyword arguments to pass to the endpoint URL.	required

Returns:

Type	Description
Any	response (httpx.Response): Response from the API request.

Raises:

Type	Description
-ImportError	If the httpx library is not installed.
-HTTPError	If the API request fails.

Examples:

>>> response = client.request("getCollaborators", "GET",
                    owner="owner", repo="repo", state="all")

Source code in llama_index/readers/github/collaborators/github_client.py

async def request(
    self,
    endpoint: str,
    method: str,
    headers: Dict[str, Any] = {},
    params: Dict[str, Any] = {},
    **kwargs: Any,
) -> Any:
    """
    Makes an API request to the GitHub API.

    Args:
        - `endpoint (str)`: Name of the endpoint to make the request to.
        - `method (str)`: HTTP method to use for the request.
        - `headers (dict)`: HTTP headers to include in the request.
        - `**kwargs`: Keyword arguments to pass to the endpoint URL.

    Returns:
        - `response (httpx.Response)`: Response from the API request.

    Raises:
        - ImportError: If the `httpx` library is not installed.
        - httpx.HTTPError: If the API request fails.

    Examples:
        >>> response = client.request("getCollaborators", "GET",
                            owner="owner", repo="repo", state="all")

    """
    try:
        import httpx
    except ImportError:
        raise ImportError(
            "`https` package not found, please run `pip install httpx`"
        )

    # Get authentication headers (may fetch fresh token for GitHub App)
    auth_headers = await self._get_auth_headers()
    _headers = {**auth_headers, **headers}

    _client: httpx.AsyncClient
    async with httpx.AsyncClient(
        headers=_headers, base_url=self._base_url, params=params
    ) as _client:
        try:
            response = await _client.request(
                method, url=self._endpoints[endpoint].format(**kwargs)
            )
            response.raise_for_status()
        except httpx.HTTPError as excp:
            print(f"HTTP Exception for {excp.request.url} - {excp}")
            raise excp  # noqa: TRY201
        return response

get_collaborators async

get_collaborators(owner: str, repo: str, page: int = 1) -> Dict

List collaborators in a repository.

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required

Returns:

Type	Description
Dict	See https://docs.github.com/en/rest/collaborators/collaborators?apiVersion=2022-11-28#list-repository-collaborators

Examples:

>>> repo_collaborators = client.get_collaborators("owner", "repo")

Source code in llama_index/readers/github/collaborators/github_client.py

async def get_collaborators(
    self,
    owner: str,
    repo: str,
    page: int = 1,
) -> Dict:
    """
    List collaborators in a repository.

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.

    Returns:
        - See https://docs.github.com/en/rest/collaborators/collaborators?apiVersion=2022-11-28#list-repository-collaborators

    Examples:
        >>> repo_collaborators = client.get_collaborators("owner", "repo")

    """
    return (
        await self.request(
            endpoint="getCollaborators",
            method="GET",
            params={
                "per_page": 100,
                "page": page,
            },
            owner=owner,
            repo=repo,
        )
    ).json()

GitHubIssuesClient

An asynchronous client for interacting with the GitHub API for issues.

The client supports two authentication methods: 1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var 2. GitHub App - passed as github_app_auth parameter

Examples:

>>> # Using Personal Access Token
>>> client = GitHubIssuesClient("my_github_token")
>>> issues = client.get_issues("owner", "repo")
>>>
>>> # Using GitHub App
>>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
>>> app_auth = GitHubAppAuth(app_id="123", private_key=key, installation_id="456")
>>> client = GitHubIssuesClient(github_app_auth=app_auth)

Source code in llama_index/readers/github/issues/github_client.py

class GitHubIssuesClient:
    """
    An asynchronous client for interacting with the GitHub API for issues.

    The client supports two authentication methods:
    1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var
    2. GitHub App - passed as github_app_auth parameter

    Examples:
        >>> # Using Personal Access Token
        >>> client = GitHubIssuesClient("my_github_token")
        >>> issues = client.get_issues("owner", "repo")
        >>>
        >>> # Using GitHub App
        >>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
        >>> app_auth = GitHubAppAuth(app_id="123", private_key=key, installation_id="456")
        >>> client = GitHubIssuesClient(github_app_auth=app_auth)

    """

    DEFAULT_BASE_URL = "https://api.github.com"
    DEFAULT_API_VERSION = "2022-11-28"

    def __init__(
        self,
        github_token: Optional[str] = None,
        github_app_auth: Optional[Union["GitHubAppAuth", Any]] = None,
        base_url: str = DEFAULT_BASE_URL,
        api_version: str = DEFAULT_API_VERSION,
        verbose: bool = False,
    ) -> None:
        """
        Initialize the GitHubIssuesClient.

        Args:
            - github_token (str, optional): GitHub token for authentication.
                If not provided, the client will try to get it from
                the GITHUB_TOKEN environment variable. Mutually exclusive with github_app_auth.
            - github_app_auth (GitHubAppAuth, optional): GitHub App authentication handler.
                Mutually exclusive with github_token.
            - base_url (str): Base URL for the GitHub API
                (defaults to "https://api.github.com").
            - api_version (str): GitHub API version (defaults to "2022-11-28").
            - verbose (bool): Whether to print verbose output (defaults to False).

        Raises:
            ValueError: If neither github_token nor github_app_auth is provided,
                       or if both are provided.

        """
        # Validate authentication parameters
        if github_token is not None and github_app_auth is not None:
            raise ValueError(
                "Cannot provide both github_token and github_app_auth. "
                "Please use only one authentication method."
            )

        self._base_url = base_url
        self._api_version = api_version
        self._verbose = verbose
        self._github_app_auth = github_app_auth
        self._github_token = None

        # Set up authentication
        if github_app_auth is not None:
            self._use_github_app = True
        else:
            self._use_github_app = False
            if github_token is None:
                github_token = os.getenv("GITHUB_TOKEN")
                if github_token is None:
                    raise ValueError(
                        "Please provide a GitHub token or GitHub App authentication. "
                        + "You can pass github_token as an argument, "
                        + "set the GITHUB_TOKEN environment variable, "
                        + "or pass github_app_auth for GitHub App authentication."
                    )
            self._github_token = github_token

        self._endpoints = {
            "getIssues": "/repos/{owner}/{repo}/issues",
        }

        # Base headers (Authorization header will be added per-request)
        self._base_headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": f"{self._api_version}",
        }

        # For backward compatibility, keep _headers with PAT token
        if not self._use_github_app:
            self._headers = {
                **self._base_headers,
                "Authorization": f"Bearer {self._github_token}",
            }
        else:
            self._headers = self._base_headers.copy()

    def get_all_endpoints(self) -> Dict[str, str]:
        """Get all available endpoints."""
        return {**self._endpoints}

    async def _get_auth_headers(self) -> Dict[str, str]:
        """Get authentication headers."""
        if self._use_github_app:
            token = await self._github_app_auth.get_installation_token()
            return {
                **self._base_headers,
                "Authorization": f"Bearer {token}",
            }
        else:
            return self._headers

    async def request(
        self,
        endpoint: str,
        method: str,
        headers: Dict[str, Any] = {},
        params: Dict[str, Any] = {},
        **kwargs: Any,
    ) -> Any:
        """
        Makes an API request to the GitHub API.

        Args:
            - `endpoint (str)`: Name of the endpoint to make the request to.
            - `method (str)`: HTTP method to use for the request.
            - `headers (dict)`: HTTP headers to include in the request.
            - `**kwargs`: Keyword arguments to pass to the endpoint URL.

        Returns:
            - `response (httpx.Response)`: Response from the API request.

        Raises:
            - ImportError: If the `httpx` library is not installed.
            - httpx.HTTPError: If the API request fails.

        Examples:
            >>> response = client.request("getIssues", "GET",
                                owner="owner", repo="repo", state="all")

        """
        try:
            import httpx
        except ImportError:
            raise ImportError(
                "`https` package not found, please run `pip install httpx`"
            )

        # Get authentication headers (may fetch fresh token for GitHub App)
        auth_headers = await self._get_auth_headers()
        _headers = {**auth_headers, **headers}

        _client: httpx.AsyncClient
        async with httpx.AsyncClient(
            headers=_headers,
            base_url=self._base_url,
            params=params,
            follow_redirects=True,
        ) as _client:
            try:
                response = await _client.request(
                    method, url=self._endpoints[endpoint].format(**kwargs)
                )
                response.raise_for_status()
            except httpx.HTTPError as excp:
                print(f"HTTP Exception for {excp.request.url} - {excp}")
                raise excp  # noqa: TRY201
            return response

    async def get_issues(
        self,
        owner: str,
        repo: str,
        state: str = "open",
        page: int = 1,
    ) -> Dict:
        """
        List issues in a repository.

        Note: GitHub's REST API considers every pull request an issue, but not every issue is a pull request.
        For this reason, "Issues" endpoints may return both issues and pull requests in the response.
        You can identify pull requests by the pull_request key.
        Be aware that the id of a pull request returned from "Issues" endpoints will be an issue id.
        To find out the pull request id, use the "List pull requests" endpoint.

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.
            - `state (str)`: Indicates the state of the issues to return.
                Default: open
                Can be one of: open, closed, all.

        Returns:
            - See https://docs.github.com/en/rest/issues/issues?apiVersion=2022-11-28#list-repository-issues

        Examples:
            >>> repo_issues = client.get_issues("owner", "repo")

        """
        return (
            await self.request(
                endpoint="getIssues",
                method="GET",
                params={
                    "state": state,
                    "per_page": 100,
                    "sort": "updated",
                    "direction": "desc",
                    "page": page,
                },
                owner=owner,
                repo=repo,
            )
        ).json()

get_all_endpoints

get_all_endpoints() -> Dict[str, str]

Get all available endpoints.

Source code in llama_index/readers/github/issues/github_client.py

def get_all_endpoints(self) -> Dict[str, str]:
    """Get all available endpoints."""
    return {**self._endpoints}

request async

request(endpoint: str, method: str, headers: Dict[str, Any] = {}, params: Dict[str, Any] = {}, **kwargs: Any) -> Any

Makes an API request to the GitHub API.

Parameters:

Name	Type	Description	Default
- `endpoint (str)`		Name of the endpoint to make the request to.	required
- `method (str)`		HTTP method to use for the request.	required
- `headers (dict)`		HTTP headers to include in the request.	required
- `**kwargs`		Keyword arguments to pass to the endpoint URL.	required

Returns:

Type	Description
Any	response (httpx.Response): Response from the API request.

Raises:

Type	Description
-ImportError	If the httpx library is not installed.
-HTTPError	If the API request fails.

Examples:

>>> response = client.request("getIssues", "GET",
                    owner="owner", repo="repo", state="all")

Source code in llama_index/readers/github/issues/github_client.py

async def request(
    self,
    endpoint: str,
    method: str,
    headers: Dict[str, Any] = {},
    params: Dict[str, Any] = {},
    **kwargs: Any,
) -> Any:
    """
    Makes an API request to the GitHub API.

    Args:
        - `endpoint (str)`: Name of the endpoint to make the request to.
        - `method (str)`: HTTP method to use for the request.
        - `headers (dict)`: HTTP headers to include in the request.
        - `**kwargs`: Keyword arguments to pass to the endpoint URL.

    Returns:
        - `response (httpx.Response)`: Response from the API request.

    Raises:
        - ImportError: If the `httpx` library is not installed.
        - httpx.HTTPError: If the API request fails.

    Examples:
        >>> response = client.request("getIssues", "GET",
                            owner="owner", repo="repo", state="all")

    """
    try:
        import httpx
    except ImportError:
        raise ImportError(
            "`https` package not found, please run `pip install httpx`"
        )

    # Get authentication headers (may fetch fresh token for GitHub App)
    auth_headers = await self._get_auth_headers()
    _headers = {**auth_headers, **headers}

    _client: httpx.AsyncClient
    async with httpx.AsyncClient(
        headers=_headers,
        base_url=self._base_url,
        params=params,
        follow_redirects=True,
    ) as _client:
        try:
            response = await _client.request(
                method, url=self._endpoints[endpoint].format(**kwargs)
            )
            response.raise_for_status()
        except httpx.HTTPError as excp:
            print(f"HTTP Exception for {excp.request.url} - {excp}")
            raise excp  # noqa: TRY201
        return response

get_issues async

get_issues(owner: str, repo: str, state: str = 'open', page: int = 1) -> Dict

List issues in a repository.

Note: GitHub's REST API considers every pull request an issue, but not every issue is a pull request. For this reason, "Issues" endpoints may return both issues and pull requests in the response. You can identify pull requests by the pull_request key. Be aware that the id of a pull request returned from "Issues" endpoints will be an issue id. To find out the pull request id, use the "List pull requests" endpoint.

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required
- `state (str)`		Indicates the state of the issues to return. Default: open Can be one of: open, closed, all.	required

Returns:

Type	Description
Dict	See https://docs.github.com/en/rest/issues/issues?apiVersion=2022-11-28#list-repository-issues

Examples:

>>> repo_issues = client.get_issues("owner", "repo")

Source code in llama_index/readers/github/issues/github_client.py

async def get_issues(
    self,
    owner: str,
    repo: str,
    state: str = "open",
    page: int = 1,
) -> Dict:
    """
    List issues in a repository.

    Note: GitHub's REST API considers every pull request an issue, but not every issue is a pull request.
    For this reason, "Issues" endpoints may return both issues and pull requests in the response.
    You can identify pull requests by the pull_request key.
    Be aware that the id of a pull request returned from "Issues" endpoints will be an issue id.
    To find out the pull request id, use the "List pull requests" endpoint.

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.
        - `state (str)`: Indicates the state of the issues to return.
            Default: open
            Can be one of: open, closed, all.

    Returns:
        - See https://docs.github.com/en/rest/issues/issues?apiVersion=2022-11-28#list-repository-issues

    Examples:
        >>> repo_issues = client.get_issues("owner", "repo")

    """
    return (
        await self.request(
            endpoint="getIssues",
            method="GET",
            params={
                "state": state,
                "per_page": 100,
                "sort": "updated",
                "direction": "desc",
                "page": page,
            },
            owner=owner,
            repo=repo,
        )
    ).json()

GitHubRepositoryIssuesReader

Bases: BaseReader

GitHub repository issues reader.

Retrieves the list of issues of a GitHub repository and returns a list of documents.

Examples:

>>> reader = GitHubRepositoryIssuesReader("owner", "repo")
>>> issues = reader.load_data()
>>> print(issues)

Source code in llama_index/readers/github/issues/base.py

class GitHubRepositoryIssuesReader(BaseReader):
    """
    GitHub repository issues reader.

    Retrieves the list of issues of a GitHub repository and returns a list of documents.

    Examples:
        >>> reader = GitHubRepositoryIssuesReader("owner", "repo")
        >>> issues = reader.load_data()
        >>> print(issues)

    """

    class IssueState(enum.Enum):
        """
        Issue type.

        Used to decide what issues to retrieve.

        Attributes:
            - OPEN: Just open issues. This is the default.
            - CLOSED: Just closed issues.
            - ALL: All issues, open and closed.

        """

        OPEN = "open"
        CLOSED = "closed"
        ALL = "all"

    class FilterType(enum.Enum):
        """
        Filter type.

        Used to determine whether the filter is inclusive or exclusive.
        """

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGitHubIssuesClient,
        owner: str,
        repo: str,
        verbose: bool = False,
    ):
        """
        Initialize params.

        Args:
            - github_client (BaseGitHubIssuesClient): GitHub client.
            - owner (str): Owner of the repository.
            - repo (str): Name of the repository.
            - verbose (bool): Whether to print verbose messages.

        Raises:
            - `ValueError`: If the github_token is not provided and
                the GITHUB_TOKEN environment variable is not set.

        """
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._verbose = verbose

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

    def load_data(
        self,
        state: Optional[IssueState] = IssueState.OPEN,
        labelFilters: Optional[List[Tuple[str, FilterType]]] = None,
    ) -> List[Document]:
        """
        Load issues from a repository and converts them to documents.

        Each issue is converted to a document by doing the following:

        - The text of the document is the concatenation of the title and the body of the issue.
        - The title of the document is the title of the issue.
        - The doc_id of the document is the issue number.
        - The extra_info of the document is a dictionary with the following keys:
            - state: State of the issue.
            - created_at: Date when the issue was created.
            - closed_at: Date when the issue was closed. Only present if the issue is closed.
            - url: URL of the issue.
            - assignee: Login of the user assigned to the issue. Only present if the issue is assigned.
        - The embedding of the document is None.
        - The doc_hash of the document is None.

        Args:
            - state (IssueState): State of the issues to retrieve. Default is IssueState.OPEN.
            - labelFilters: an optional list of filters to apply to the issue list based on labels.

        :return: list of documents

        """
        documents = []
        page = 1
        # Loop until there are no more issues
        while True:
            issues: Dict = self._loop.run_until_complete(
                self._github_client.get_issues(
                    self._owner, self._repo, state=state.value, page=page
                )
            )

            if len(issues) == 0:
                print_if_verbose(self._verbose, "No more issues found, stopping")

                break
            print_if_verbose(
                self._verbose, f"Found {len(issues)} issues in the repo page {page}"
            )
            page += 1
            filterCount = 0
            for issue in issues:
                if not self._must_include(labelFilters, issue):
                    filterCount += 1
                    continue
                title = issue["title"]
                body = issue["body"]
                document = Document(
                    doc_id=str(issue["number"]),
                    text=f"{title}\n{body}",
                )
                extra_info = {
                    "state": issue["state"],
                    "created_at": issue["created_at"],
                    # url is the API URL
                    "url": issue["url"],
                    # source is the HTML URL, more convenient for humans
                    "source": issue["html_url"],
                }
                if issue["closed_at"] is not None:
                    extra_info["closed_at"] = issue["closed_at"]
                if issue["assignee"] is not None:
                    extra_info["assignee"] = issue["assignee"]["login"]
                if issue["labels"] is not None:
                    extra_info["labels"] = [label["name"] for label in issue["labels"]]
                document.extra_info = extra_info
                documents.append(document)

            print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")
            if labelFilters is not None:
                print_if_verbose(self._verbose, f"Filtered out {filterCount} issues")

        return documents

    def _must_include(self, labelFilters, issue):
        if labelFilters is None:
            return True
        labels = [label["name"] for label in issue["labels"]]
        for labelFilter in labelFilters:
            label = labelFilter[0]
            filterType = labelFilter[1]
            # Only include issues with the label and value
            if filterType == self.FilterType.INCLUDE:
                return label in labels
            elif filterType == self.FilterType.EXCLUDE:
                return label not in labels

        return True

IssueState

Bases: Enum

Issue type.

Used to decide what issues to retrieve.

Attributes:

Name	Type	Description
-	OPEN	Just open issues. This is the default.
-	CLOSED	Just closed issues.
-	ALL	All issues, open and closed.

Source code in llama_index/readers/github/issues/base.py

class IssueState(enum.Enum):
    """
    Issue type.

    Used to decide what issues to retrieve.

    Attributes:
        - OPEN: Just open issues. This is the default.
        - CLOSED: Just closed issues.
        - ALL: All issues, open and closed.

    """

    OPEN = "open"
    CLOSED = "closed"
    ALL = "all"

FilterType

Bases: Enum

Filter type.

Used to determine whether the filter is inclusive or exclusive.

Source code in llama_index/readers/github/issues/base.py

class FilterType(enum.Enum):
    """
    Filter type.

    Used to determine whether the filter is inclusive or exclusive.
    """

    EXCLUDE = enum.auto()
    INCLUDE = enum.auto()

load_data

load_data(state: Optional[IssueState] = OPEN, labelFilters: Optional[List[Tuple[str, FilterType]]] = None) -> List[Document]

Load issues from a repository and converts them to documents.

Each issue is converted to a document by doing the following:

• The text of the document is the concatenation of the title and the body of the issue.
• The title of the document is the title of the issue.
• The doc_id of the document is the issue number.
• The extra_info of the document is a dictionary with the following keys:
  • state: State of the issue.
  • created_at: Date when the issue was created.
  • closed_at: Date when the issue was closed. Only present if the issue is closed.
  • url: URL of the issue.
  • assignee: Login of the user assigned to the issue. Only present if the issue is assigned.
• The embedding of the document is None.
• The doc_hash of the document is None.

Parameters:

Name	Type	Description	Default
- state	IssueState	State of the issues to retrieve. Default is IssueState.OPEN.	required
- labelFilters		an optional list of filters to apply to the issue list based on labels.	required

:return: list of documents

Source code in llama_index/readers/github/issues/base.py

def load_data(
    self,
    state: Optional[IssueState] = IssueState.OPEN,
    labelFilters: Optional[List[Tuple[str, FilterType]]] = None,
) -> List[Document]:
    """
    Load issues from a repository and converts them to documents.

    Each issue is converted to a document by doing the following:

    - The text of the document is the concatenation of the title and the body of the issue.
    - The title of the document is the title of the issue.
    - The doc_id of the document is the issue number.
    - The extra_info of the document is a dictionary with the following keys:
        - state: State of the issue.
        - created_at: Date when the issue was created.
        - closed_at: Date when the issue was closed. Only present if the issue is closed.
        - url: URL of the issue.
        - assignee: Login of the user assigned to the issue. Only present if the issue is assigned.
    - The embedding of the document is None.
    - The doc_hash of the document is None.

    Args:
        - state (IssueState): State of the issues to retrieve. Default is IssueState.OPEN.
        - labelFilters: an optional list of filters to apply to the issue list based on labels.

    :return: list of documents

    """
    documents = []
    page = 1
    # Loop until there are no more issues
    while True:
        issues: Dict = self._loop.run_until_complete(
            self._github_client.get_issues(
                self._owner, self._repo, state=state.value, page=page
            )
        )

        if len(issues) == 0:
            print_if_verbose(self._verbose, "No more issues found, stopping")

            break
        print_if_verbose(
            self._verbose, f"Found {len(issues)} issues in the repo page {page}"
        )
        page += 1
        filterCount = 0
        for issue in issues:
            if not self._must_include(labelFilters, issue):
                filterCount += 1
                continue
            title = issue["title"]
            body = issue["body"]
            document = Document(
                doc_id=str(issue["number"]),
                text=f"{title}\n{body}",
            )
            extra_info = {
                "state": issue["state"],
                "created_at": issue["created_at"],
                # url is the API URL
                "url": issue["url"],
                # source is the HTML URL, more convenient for humans
                "source": issue["html_url"],
            }
            if issue["closed_at"] is not None:
                extra_info["closed_at"] = issue["closed_at"]
            if issue["assignee"] is not None:
                extra_info["assignee"] = issue["assignee"]["login"]
            if issue["labels"] is not None:
                extra_info["labels"] = [label["name"] for label in issue["labels"]]
            document.extra_info = extra_info
            documents.append(document)

        print_if_verbose(self._verbose, f"Resulted in {len(documents)} documents")
        if labelFilters is not None:
            print_if_verbose(self._verbose, f"Filtered out {filterCount} issues")

    return documents

GithubClient

An asynchronous client for interacting with the Github API.

This client is used for making API requests to Github. It provides methods for accessing the Github API endpoints. The client supports two authentication methods: 1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var 2. GitHub App - passed as github_app_auth parameter

Examples:

>>> # Using Personal Access Token
>>> client = GithubClient("my_github_token")
>>> branch_info = client.get_branch("owner", "repo", "branch")
>>>
>>> # Using GitHub App
>>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
>>> with open("private-key.pem", "r") as f:
...     private_key = f.read()
>>> app_auth = GitHubAppAuth(
...     app_id="123456",
...     private_key=private_key,
...     installation_id="789012"
... )
>>> client = GithubClient(github_app_auth=app_auth)

Source code in llama_index/readers/github/repository/github_client.py

class GithubClient:
    """
    An asynchronous client for interacting with the Github API.

    This client is used for making API requests to Github.
    It provides methods for accessing the Github API endpoints.
    The client supports two authentication methods:
    1. Personal Access Token (PAT) - passed as github_token or via GITHUB_TOKEN env var
    2. GitHub App - passed as github_app_auth parameter

    Examples:
        >>> # Using Personal Access Token
        >>> client = GithubClient("my_github_token")
        >>> branch_info = client.get_branch("owner", "repo", "branch")
        >>>
        >>> # Using GitHub App
        >>> from llama_index.readers.github.github_app_auth import GitHubAppAuth
        >>> with open("private-key.pem", "r") as f:
        ...     private_key = f.read()
        >>> app_auth = GitHubAppAuth(
        ...     app_id="123456",
        ...     private_key=private_key,
        ...     installation_id="789012"
        ... )
        >>> client = GithubClient(github_app_auth=app_auth)

    """

    DEFAULT_BASE_URL = "https://api.github.com"
    DEFAULT_API_VERSION = "2022-11-28"

    def __init__(
        self,
        github_token: Optional[str] = None,
        github_app_auth: Optional[Union["GitHubAppAuth", Any]] = None,
        base_url: str = DEFAULT_BASE_URL,
        api_version: str = DEFAULT_API_VERSION,
        verbose: bool = False,
        fail_on_http_error: bool = True,
    ) -> None:
        """
        Initialize the GithubClient.

        Args:
            - github_token (str, optional): Github token for authentication.
                If not provided, the client will try to get it from
                the GITHUB_TOKEN environment variable. Mutually exclusive with github_app_auth.
            - github_app_auth (GitHubAppAuth, optional): GitHub App authentication handler.
                Mutually exclusive with github_token.
            - base_url (str): Base URL for the Github API
                (defaults to "https://api.github.com").
            - api_version (str): Github API version (defaults to "2022-11-28").
            - verbose (bool): Whether to print verbose output (defaults to False).
            - fail_on_http_error (bool): Whether to raise an exception on HTTP errors (defaults to True).

        Raises:
            ValueError: If neither github_token nor github_app_auth is provided,
                       or if both are provided.

        """
        # Validate authentication parameters
        if github_token is not None and github_app_auth is not None:
            raise ValueError(
                "Cannot provide both github_token and github_app_auth. "
                "Please use only one authentication method."
            )

        self._base_url = base_url
        self._api_version = api_version
        self._verbose = verbose
        self._fail_on_http_error = fail_on_http_error
        self._github_app_auth = github_app_auth
        self._github_token = None

        # Set up authentication
        if github_app_auth is not None:
            # Using GitHub App authentication
            self._use_github_app = True
        else:
            # Using PAT authentication
            self._use_github_app = False
            if github_token is None:
                github_token = os.getenv("GITHUB_TOKEN")
                if github_token is None:
                    raise ValueError(
                        "Please provide a Github token or GitHub App authentication. "
                        + "You can pass github_token as an argument, "
                        + "set the GITHUB_TOKEN environment variable, "
                        + "or pass github_app_auth for GitHub App authentication."
                    )
            self._github_token = github_token

        self._endpoints = {
            "getTree": "/repos/{owner}/{repo}/git/trees/{tree_sha}",
            "getBranch": "/repos/{owner}/{repo}/branches/{branch}",
            "getBlob": "/repos/{owner}/{repo}/git/blobs/{file_sha}",
            "getCommit": "/repos/{owner}/{repo}/commits/{commit_sha}",
        }

        # Base headers (Authorization header will be added per-request)
        self._base_headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": f"{self._api_version}",
        }

        # For backward compatibility, keep _headers with PAT token
        if not self._use_github_app:
            self._headers = {
                **self._base_headers,
                "Authorization": f"Bearer {self._github_token}",
            }
        else:
            # Headers will be generated per-request for GitHub App
            self._headers = self._base_headers.copy()

    def get_all_endpoints(self) -> Dict[str, str]:
        """Get all available endpoints."""
        return {**self._endpoints}

    async def _get_auth_headers(self) -> Dict[str, str]:
        """
        Get authentication headers.

        For PAT authentication, returns cached headers.
        For GitHub App authentication, fetches a fresh installation token if needed.

        Returns:
            Dictionary containing authentication headers.

        """
        if self._use_github_app:
            # Get fresh token from GitHub App auth
            token = await self._github_app_auth.get_installation_token()
            return {
                **self._base_headers,
                "Authorization": f"Bearer {token}",
            }
        else:
            # Return cached headers with PAT
            return self._headers

    async def request(
        self,
        endpoint: str,
        method: str,
        headers: Dict[str, Any] = {},
        timeout: Optional[int] = 5,
        retries: int = 0,
        **kwargs: Any,
    ) -> Any:
        """
        Make an API request to the Github API.

        This method is used for making API requests to the Github API.
        It is used internally by the other methods in the client.

        Args:
            - `endpoint (str)`: Name of the endpoint to make the request to.
            - `method (str)`: HTTP method to use for the request.
            - `headers (dict)`: HTTP headers to include in the request.
            - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
            - `retries (int)`: Number of retries for the request. Default is 0.
            - `**kwargs`: Keyword arguments to pass to the endpoint URL.

        Returns:
            - `response (httpx.Response)`: Response from the API request.

        Raises:
            - ImportError: If the `httpx` library is not installed.
            - httpx.HTTPError: If the API request fails and fail_on_http_error is True.

        Examples:
            >>> response = client.request("getTree", "GET",
                                owner="owner", repo="repo",
                                tree_sha="tree_sha", timeout=5, retries=0)

        """
        try:
            import httpx
        except ImportError:
            raise ImportError(
                "Please install httpx to use the GithubRepositoryReader. "
                "You can do so by running `pip install httpx`."
            )

        # Get authentication headers (may fetch fresh token for GitHub App)
        auth_headers = await self._get_auth_headers()
        _headers = {**auth_headers, **headers}

        _client: httpx.AsyncClient
        async with httpx.AsyncClient(
            headers=_headers,
            base_url=self._base_url,
            timeout=timeout,
            transport=httpx.AsyncHTTPTransport(retries=retries),
        ) as _client:
            try:
                response = await _client.request(
                    method, url=self._endpoints[endpoint].format(**kwargs)
                )
            except httpx.HTTPError as excp:
                print(f"HTTP Exception for {excp.request.url} - {excp}")
                raise excp  # noqa: TRY201
            return response

    async def get_branch(
        self,
        owner: str,
        repo: str,
        branch: Optional[str] = None,
        branch_name: Optional[str] = None,
        timeout: Optional[int] = 5,
        retries: int = 0,
    ) -> GitBranchResponseModel:
        """
        Get information about a branch. (Github API endpoint: getBranch).

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.
            - `branch (str)`: Name of the branch.
            - `branch_name (str)`: Name of the branch (alternative to `branch`).
            - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
            - `retries (int)`: Number of retries for the request. Default is 0.

        Returns:
            - `branch_info (GitBranchResponseModel)`: Information about the branch.

        Examples:
            >>> branch_info = client.get_branch("owner", "repo", "branch")

        """
        if branch is None:
            if branch_name is None:
                raise ValueError("Either branch or branch_name must be provided.")
            branch = branch_name

        return GitBranchResponseModel.from_json(
            (
                await self.request(
                    "getBranch",
                    "GET",
                    owner=owner,
                    repo=repo,
                    branch=branch,
                    timeout=timeout,
                    retries=retries,
                )
            ).text
        )

    async def get_tree(
        self,
        owner: str,
        repo: str,
        tree_sha: str,
        timeout: Optional[int] = 5,
        retries: int = 0,
    ) -> GitTreeResponseModel:
        """
        Get information about a tree. (Github API endpoint: getTree).

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.
            - `tree_sha (str)`: SHA of the tree.
            - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
            - `retries (int)`: Number of retries for the request. Default is 0.

        Returns:
            - `tree_info (GitTreeResponseModel)`: Information about the tree.

        Examples:
            >>> tree_info = client.get_tree("owner", "repo", "tree_sha")

        """
        return GitTreeResponseModel.from_json(
            (
                await self.request(
                    "getTree",
                    "GET",
                    owner=owner,
                    repo=repo,
                    tree_sha=tree_sha,
                    timeout=timeout,
                    retries=retries,
                )
            ).text
        )

    async def get_blob(
        self,
        owner: str,
        repo: str,
        file_sha: str,
        timeout: Optional[int] = 5,
        retries: int = 0,
    ) -> Optional[GitBlobResponseModel]:
        """
        Get information about a blob. (Github API endpoint: getBlob).

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.
            - `file_sha (str)`: SHA of the file.
            - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
            - `retries (int)`: Number of retries for the request. Default is 0.

        Returns:
            - `blob_info (GitBlobResponseModel)`: Information about the blob.

        Examples:
            >>> blob_info = client.get_blob("owner", "repo", "file_sha")

        """
        try:
            return GitBlobResponseModel.from_json(
                (
                    await self.request(
                        "getBlob",
                        "GET",
                        owner=owner,
                        repo=repo,
                        file_sha=file_sha,
                        timeout=timeout,
                        retries=retries,
                    )
                ).text
            )
        except KeyError:
            print(f"Failed to get blob for {owner}/{repo}/{file_sha}")
            return None
        except HTTPError as excp:
            print(f"HTTP Exception for {excp.request.url} - {excp}")
            if self._fail_on_http_error:
                raise
            else:
                return None

    async def get_commit(
        self,
        owner: str,
        repo: str,
        commit_sha: str,
        timeout: Optional[int] = 5,
        retries: int = 0,
    ) -> GitCommitResponseModel:
        """
        Get information about a commit. (Github API endpoint: getCommit).

        Args:
            - `owner (str)`: Owner of the repository.
            - `repo (str)`: Name of the repository.
            - `commit_sha (str)`: SHA of the commit.
            - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
            - `retries (int)`: Number of retries for the request. Default is 0.

        Returns:
            - `commit_info (GitCommitResponseModel)`: Information about the commit.

        Examples:
            >>> commit_info = client.get_commit("owner", "repo", "commit_sha")

        """
        return GitCommitResponseModel.from_json(
            (
                await self.request(
                    "getCommit",
                    "GET",
                    owner=owner,
                    repo=repo,
                    commit_sha=commit_sha,
                    timeout=timeout,
                    retries=retries,
                )
            ).text
        )

get_all_endpoints

get_all_endpoints() -> Dict[str, str]

Get all available endpoints.

Source code in llama_index/readers/github/repository/github_client.py

def get_all_endpoints(self) -> Dict[str, str]:
    """Get all available endpoints."""
    return {**self._endpoints}

request async

request(endpoint: str, method: str, headers: Dict[str, Any] = {}, timeout: Optional[int] = 5, retries: int = 0, **kwargs: Any) -> Any

Make an API request to the Github API.

This method is used for making API requests to the Github API. It is used internally by the other methods in the client.

Parameters:

Name	Type	Description	Default
- `endpoint (str)`		Name of the endpoint to make the request to.	required
- `method (str)`		HTTP method to use for the request.	required
- `headers (dict)`		HTTP headers to include in the request.	required
- `timeout (int or None)`		Timeout for the request in seconds. Default is 5.	required
- `retries (int)`		Number of retries for the request. Default is 0.	required
- `**kwargs`		Keyword arguments to pass to the endpoint URL.	required

Returns:

Type	Description
Any	response (httpx.Response): Response from the API request.

Raises:

Type	Description
-ImportError	If the httpx library is not installed.
-HTTPError	If the API request fails and fail_on_http_error is True.

Examples:

>>> response = client.request("getTree", "GET",
                    owner="owner", repo="repo",
                    tree_sha="tree_sha", timeout=5, retries=0)

Source code in llama_index/readers/github/repository/github_client.py

async def request(
    self,
    endpoint: str,
    method: str,
    headers: Dict[str, Any] = {},
    timeout: Optional[int] = 5,
    retries: int = 0,
    **kwargs: Any,
) -> Any:
    """
    Make an API request to the Github API.

    This method is used for making API requests to the Github API.
    It is used internally by the other methods in the client.

    Args:
        - `endpoint (str)`: Name of the endpoint to make the request to.
        - `method (str)`: HTTP method to use for the request.
        - `headers (dict)`: HTTP headers to include in the request.
        - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
        - `retries (int)`: Number of retries for the request. Default is 0.
        - `**kwargs`: Keyword arguments to pass to the endpoint URL.

    Returns:
        - `response (httpx.Response)`: Response from the API request.

    Raises:
        - ImportError: If the `httpx` library is not installed.
        - httpx.HTTPError: If the API request fails and fail_on_http_error is True.

    Examples:
        >>> response = client.request("getTree", "GET",
                            owner="owner", repo="repo",
                            tree_sha="tree_sha", timeout=5, retries=0)

    """
    try:
        import httpx
    except ImportError:
        raise ImportError(
            "Please install httpx to use the GithubRepositoryReader. "
            "You can do so by running `pip install httpx`."
        )

    # Get authentication headers (may fetch fresh token for GitHub App)
    auth_headers = await self._get_auth_headers()
    _headers = {**auth_headers, **headers}

    _client: httpx.AsyncClient
    async with httpx.AsyncClient(
        headers=_headers,
        base_url=self._base_url,
        timeout=timeout,
        transport=httpx.AsyncHTTPTransport(retries=retries),
    ) as _client:
        try:
            response = await _client.request(
                method, url=self._endpoints[endpoint].format(**kwargs)
            )
        except httpx.HTTPError as excp:
            print(f"HTTP Exception for {excp.request.url} - {excp}")
            raise excp  # noqa: TRY201
        return response

get_branch async

get_branch(owner: str, repo: str, branch: Optional[str] = None, branch_name: Optional[str] = None, timeout: Optional[int] = 5, retries: int = 0) -> GitBranchResponseModel

Get information about a branch. (Github API endpoint: getBranch).

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required
- `branch (str)`		Name of the branch.	required
- `branch_name (str)`		Name of the branch (alternative to branch).	required
- `timeout (int or None)`		Timeout for the request in seconds. Default is 5.	required
- `retries (int)`		Number of retries for the request. Default is 0.	required

Returns:

Type	Description
GitBranchResponseModel	branch_info (GitBranchResponseModel): Information about the branch.

Examples:

>>> branch_info = client.get_branch("owner", "repo", "branch")

Source code in llama_index/readers/github/repository/github_client.py

async def get_branch(
    self,
    owner: str,
    repo: str,
    branch: Optional[str] = None,
    branch_name: Optional[str] = None,
    timeout: Optional[int] = 5,
    retries: int = 0,
) -> GitBranchResponseModel:
    """
    Get information about a branch. (Github API endpoint: getBranch).

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.
        - `branch (str)`: Name of the branch.
        - `branch_name (str)`: Name of the branch (alternative to `branch`).
        - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
        - `retries (int)`: Number of retries for the request. Default is 0.

    Returns:
        - `branch_info (GitBranchResponseModel)`: Information about the branch.

    Examples:
        >>> branch_info = client.get_branch("owner", "repo", "branch")

    """
    if branch is None:
        if branch_name is None:
            raise ValueError("Either branch or branch_name must be provided.")
        branch = branch_name

    return GitBranchResponseModel.from_json(
        (
            await self.request(
                "getBranch",
                "GET",
                owner=owner,
                repo=repo,
                branch=branch,
                timeout=timeout,
                retries=retries,
            )
        ).text
    )

get_tree async

get_tree(owner: str, repo: str, tree_sha: str, timeout: Optional[int] = 5, retries: int = 0) -> GitTreeResponseModel

Get information about a tree. (Github API endpoint: getTree).

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required
- `tree_sha (str)`		SHA of the tree.	required
- `timeout (int or None)`		Timeout for the request in seconds. Default is 5.	required
- `retries (int)`		Number of retries for the request. Default is 0.	required

Returns:

Type	Description
GitTreeResponseModel	tree_info (GitTreeResponseModel): Information about the tree.

Examples:

>>> tree_info = client.get_tree("owner", "repo", "tree_sha")

Source code in llama_index/readers/github/repository/github_client.py

async def get_tree(
    self,
    owner: str,
    repo: str,
    tree_sha: str,
    timeout: Optional[int] = 5,
    retries: int = 0,
) -> GitTreeResponseModel:
    """
    Get information about a tree. (Github API endpoint: getTree).

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.
        - `tree_sha (str)`: SHA of the tree.
        - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
        - `retries (int)`: Number of retries for the request. Default is 0.

    Returns:
        - `tree_info (GitTreeResponseModel)`: Information about the tree.

    Examples:
        >>> tree_info = client.get_tree("owner", "repo", "tree_sha")

    """
    return GitTreeResponseModel.from_json(
        (
            await self.request(
                "getTree",
                "GET",
                owner=owner,
                repo=repo,
                tree_sha=tree_sha,
                timeout=timeout,
                retries=retries,
            )
        ).text
    )

get_blob async

get_blob(owner: str, repo: str, file_sha: str, timeout: Optional[int] = 5, retries: int = 0) -> Optional[GitBlobResponseModel]

Get information about a blob. (Github API endpoint: getBlob).

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required
- `file_sha (str)`		SHA of the file.	required
- `timeout (int or None)`		Timeout for the request in seconds. Default is 5.	required
- `retries (int)`		Number of retries for the request. Default is 0.	required

Returns:

Type	Description
Optional[GitBlobResponseModel]	blob_info (GitBlobResponseModel): Information about the blob.

Examples:

>>> blob_info = client.get_blob("owner", "repo", "file_sha")

Source code in llama_index/readers/github/repository/github_client.py

async def get_blob(
    self,
    owner: str,
    repo: str,
    file_sha: str,
    timeout: Optional[int] = 5,
    retries: int = 0,
) -> Optional[GitBlobResponseModel]:
    """
    Get information about a blob. (Github API endpoint: getBlob).

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.
        - `file_sha (str)`: SHA of the file.
        - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
        - `retries (int)`: Number of retries for the request. Default is 0.

    Returns:
        - `blob_info (GitBlobResponseModel)`: Information about the blob.

    Examples:
        >>> blob_info = client.get_blob("owner", "repo", "file_sha")

    """
    try:
        return GitBlobResponseModel.from_json(
            (
                await self.request(
                    "getBlob",
                    "GET",
                    owner=owner,
                    repo=repo,
                    file_sha=file_sha,
                    timeout=timeout,
                    retries=retries,
                )
            ).text
        )
    except KeyError:
        print(f"Failed to get blob for {owner}/{repo}/{file_sha}")
        return None
    except HTTPError as excp:
        print(f"HTTP Exception for {excp.request.url} - {excp}")
        if self._fail_on_http_error:
            raise
        else:
            return None

get_commit async

get_commit(owner: str, repo: str, commit_sha: str, timeout: Optional[int] = 5, retries: int = 0) -> GitCommitResponseModel

Get information about a commit. (Github API endpoint: getCommit).

Parameters:

Name	Type	Description	Default
- `owner (str)`		Owner of the repository.	required
- `repo (str)`		Name of the repository.	required
- `commit_sha (str)`		SHA of the commit.	required
- `timeout (int or None)`		Timeout for the request in seconds. Default is 5.	required
- `retries (int)`		Number of retries for the request. Default is 0.	required

Returns:

Type	Description
GitCommitResponseModel	commit_info (GitCommitResponseModel): Information about the commit.

Examples:

>>> commit_info = client.get_commit("owner", "repo", "commit_sha")

Source code in llama_index/readers/github/repository/github_client.py

async def get_commit(
    self,
    owner: str,
    repo: str,
    commit_sha: str,
    timeout: Optional[int] = 5,
    retries: int = 0,
) -> GitCommitResponseModel:
    """
    Get information about a commit. (Github API endpoint: getCommit).

    Args:
        - `owner (str)`: Owner of the repository.
        - `repo (str)`: Name of the repository.
        - `commit_sha (str)`: SHA of the commit.
        - `timeout (int or None)`: Timeout for the request in seconds. Default is 5.
        - `retries (int)`: Number of retries for the request. Default is 0.

    Returns:
        - `commit_info (GitCommitResponseModel)`: Information about the commit.

    Examples:
        >>> commit_info = client.get_commit("owner", "repo", "commit_sha")

    """
    return GitCommitResponseModel.from_json(
        (
            await self.request(
                "getCommit",
                "GET",
                owner=owner,
                repo=repo,
                commit_sha=commit_sha,
                timeout=timeout,
                retries=retries,
            )
        ).text
    )

GithubRepositoryReader

Bases: BaseReader

Github repository reader.

Retrieves the contents of a Github repository and returns a list of documents. The documents are either the contents of the files in the repository or the text extracted from the files using the parser.

Examples:

>>> from llama_index.core.instrumentation import get_dispatcher
>>> from llama_index.core.instrumentation.event_handlers import BaseEventHandler
>>>
>>> # Set up event handler
>>> class GitHubEventHandler(BaseEventHandler):
...     def handle(self, event):
...         if isinstance(event, GitHubFileProcessedEvent):
...             print(f"Processed file: {event.file_path}")
...
>>> dispatcher = get_dispatcher()
>>> handler = GitHubEventHandler()
>>> dispatcher.add_event_handler(handler)
>>>
>>> client = github_client = GithubClient(
...    github_token=os.environ["GITHUB_TOKEN"],
...    verbose=True
... )
>>> reader = GithubRepositoryReader(
...    github_client=github_client,
...    owner="run-llama",
...    repo="llama_index",
... )
>>> # Load all files from a branch
>>> branch_documents = reader.load_data(branch="main")
>>> # Load all files from a commit
>>> commit_documents = reader.load_data(commit_sha="commit_sha")

Source code in llama_index/readers/github/repository/base.py

class GithubRepositoryReader(BaseReader):
    """
    Github repository reader.

    Retrieves the contents of a Github repository and returns a list of documents.
    The documents are either the contents of the files in the repository or the text
    extracted from the files using the parser.

    Examples:
        >>> from llama_index.core.instrumentation import get_dispatcher
        >>> from llama_index.core.instrumentation.event_handlers import BaseEventHandler
        >>>
        >>> # Set up event handler
        >>> class GitHubEventHandler(BaseEventHandler):
        ...     def handle(self, event):
        ...         if isinstance(event, GitHubFileProcessedEvent):
        ...             print(f"Processed file: {event.file_path}")
        ...
        >>> dispatcher = get_dispatcher()
        >>> handler = GitHubEventHandler()
        >>> dispatcher.add_event_handler(handler)
        >>>
        >>> client = github_client = GithubClient(
        ...    github_token=os.environ["GITHUB_TOKEN"],
        ...    verbose=True
        ... )
        >>> reader = GithubRepositoryReader(
        ...    github_client=github_client,
        ...    owner="run-llama",
        ...    repo="llama_index",
        ... )
        >>> # Load all files from a branch
        >>> branch_documents = reader.load_data(branch="main")
        >>> # Load all files from a commit
        >>> commit_documents = reader.load_data(commit_sha="commit_sha")

    """

    class FilterType(enum.Enum):
        """
        Filter type.

        Used to determine whether the filter is inclusive or exclusive.

        Attributes:
            - EXCLUDE: Exclude the files in the directories or with the extensions.
            - INCLUDE: Include only the files in the directories or with the extensions.

        """

        EXCLUDE = enum.auto()
        INCLUDE = enum.auto()

    def __init__(
        self,
        github_client: BaseGithubClient,
        owner: str,
        repo: str,
        use_parser: bool = False,
        verbose: bool = False,
        concurrent_requests: int = 5,
        timeout: Optional[int] = 5,
        retries: int = 0,
        filter_directories: Optional[Tuple[List[str], FilterType]] = None,
        filter_file_extensions: Optional[Tuple[List[str], FilterType]] = None,
        filter_file_paths: Optional[Tuple[List[str], FilterType]] = None,
        custom_parsers: Optional[Dict[str, BaseReader]] = None,
        process_file_callback: Optional[Callable[[str, int], tuple[bool, str]]] = None,
        custom_folder: Optional[str] = None,
        logger: Optional[logging.Logger] = None,
        fail_on_error: bool = True,
    ):
        """
        Initialize params.

        Args:
            - github_client (BaseGithubClient): Github client.
            - owner (str): Owner of the repository.
            - repo (str): Name of the repository.
            - use_parser (bool): Whether to use the parser to extract
                the text from the files.
            - verbose (bool): Whether to print verbose messages.
            - concurrent_requests (int): Number of concurrent requests to
                make to the Github API.
            - timeout (int or None): Timeout for the requests to the Github API. Default is 5.
            - retries (int): Number of retries for requests made to the Github API. Default is 0.
              This limit applies individually to each request made by this class.
            - filter_directories (Optional[Tuple[List[str], FilterType]]): Tuple
                containing a list of directories and a FilterType. If the FilterType
                is INCLUDE, only the files in the directories in the list will be
                included. If the FilterType is EXCLUDE, the files in the directories
                in the list will be excluded.
            - filter_file_extensions (Optional[Tuple[List[str], FilterType]]): Tuple
                containing a list of file extensions and a FilterType. If the
                FilterType is INCLUDE, only the files with the extensions in the list
                will be included. If the FilterType is EXCLUDE, the files with the
                extensions in the list will be excluded.
            - custom_parsers (Optional[Dict[str, BaseReader]]): Dictionary mapping
                file types to custom parsers for processing specific file extensions.
            - process_file_callback (Optional[Callable[[str, int], tuple[bool, str]]]): Callback function
                to determine if a file should be processed. Takes file_path and file_size
                Returns a tuple of (should_process: bool, reason: str).
            - custom_folder (Optional[str]): Custom folder path for temporary file storage in custom parsers.
                Defaults to current working directory.
            - logger (Optional[logging.Logger]): Custom logger instance.
            - fail_on_error (bool): Whether to raise exceptions on processing errors.
                If False, errors are logged and processing continues.

        Note:
            The reader uses LlamaIndex's instrumentation system for event notifications.
            Events are automatically dispatched to any registered event handlers via the global dispatcher.
            Available events include:
            - GitHubRepositoryProcessingStartedEvent
            - GitHubRepositoryProcessingCompletedEvent
            - GitHubTotalFilesToProcessEvent
            - GitHubFileProcessingStartedEvent
            - GitHubFileProcessedEvent
            - GitHubFileSkippedEvent
            - GitHubFileFailedEvent

        Raises:
            - `ValueError`: If the github_token is not provided and
                the GITHUB_TOKEN environment variable is not set.

        """
        super().__init__()

        self._owner = owner
        self._repo = repo
        self._use_parser = use_parser
        self._verbose = verbose
        self._concurrent_requests = concurrent_requests
        self._timeout = timeout
        self._retries = retries
        self._filter_directories = filter_directories
        self._filter_file_extensions = filter_file_extensions
        self._filter_file_paths = filter_file_paths
        self.custom_folder = custom_folder or os.getcwd()
        self.logger = logger or logging.getLogger(__name__)
        self._process_file_callback = process_file_callback
        self.fail_on_error = fail_on_error

        # Set up the event loop
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            # If there is no running loop, create a new one
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)

        self._github_client = github_client

        self._file_readers: Dict[str, BaseReader] = custom_parsers or {}
        self._supported_suffix = list(DEFAULT_FILE_READER_CLS.keys())
        self.dispatcher = get_dispatcher()

    def _check_filter_directories(self, tree_obj_path: str) -> bool:
        """
        Check if a tree object should be allowed based on the directories.

        :param `tree_obj_path`: path of the tree object i.e. 'llama_index/readers'

        :return: True if the tree object should be allowed, False otherwise
        """
        if self._filter_directories is None:
            return True
        filter_directories, filter_type = self._filter_directories
        print_if_verbose(
            self._verbose,
            f"Checking {tree_obj_path} whether to {filter_type} it"
            + f" based on the filter directories: {filter_directories}",
        )

        if filter_type == self.FilterType.EXCLUDE:
            print_if_verbose(
                self._verbose,
                f"Checking if {tree_obj_path} is not a subdirectory of any of the"
                " filter directories",
            )
            return not any(
                tree_obj_path.startswith(directory) for directory in filter_directories
            )
        if filter_type == self.FilterType.INCLUDE:
            print_if_verbose(
                self._verbose,
                f"Checking if {tree_obj_path} is a subdirectory of any of the filter"
                " directories",
            )
            return any(
                tree_obj_path.startswith(directory)
                or directory.startswith(tree_obj_path)
                for directory in filter_directories
            )
        raise ValueError(
            f"Unknown filter type: {filter_type}. "
            "Please use either 'INCLUDE' or 'EXCLUDE'."
        )

    def _check_filter_file_paths(self, tree_obj_path: str) -> bool:
        """
        Check if a tree object should be allowed based on the file paths.

        :param `tree_obj_path`: path of the tree object i.e. 'llama_index/readers'

        :return: True if the tree object should be allowed, False otherwise
        """
        if self._filter_file_paths is None:
            return True
        filter_file_paths, filter_type = self._filter_file_paths
        print_if_verbose(
            self._verbose,
            f"Checking {tree_obj_path} whether to {filter_type} it"
            + f" based on the filter file paths: {filter_file_paths}",
        )

        if filter_type == self.FilterType.EXCLUDE:
            to_exclude = tree_obj_path not in filter_file_paths
            file_extension = get_file_extension(tree_obj_path)
            reason = "File in Exclude List"
            if to_exclude:
                self.logger.info(f"Skipping file {tree_obj_path}: {reason}")
                self.dispatcher.event(
                    GitHubFileSkippedEvent(
                        file_path=tree_obj_path,
                        file_type=file_extension,
                        reason=reason,
                    )
                )
            return to_exclude
        if filter_type == self.FilterType.INCLUDE:
            return tree_obj_path in filter_file_paths
        raise ValueError(
            f"Unknown filter type: {filter_type}. "
            "Please use either 'INCLUDE' or 'EXCLUDE'."
        )

    def _check_user_callback(self, full_path: str, file_size: int) -> bool:
        """
        Determine if a file should be processed based on the callback function.

        :param full_path: Full path to the file
        :param file_size: Size of the file in bytes
        :return: True if file should be processed, False otherwise
        """
        if not self._process_file_callback:
            return True

        file_extension = get_file_extension(full_path)

        try:
            should_process, reason = self._process_file_callback(full_path, file_size)
        except Exception as e:
            self.logger.error(f"Error in process_file_callback for {full_path}: {e}")
            if self.fail_on_error:
                raise
            # If fail_on_error is False, skip the file
            reason = f"Callback error: {e!s}"
            should_process = False

        if not should_process:
            self.logger.info(f"Skipping file {full_path}: {reason}")
            self.dispatcher.event(
                GitHubFileSkippedEvent(
                    file_path=full_path,
                    file_type=file_extension,
                    reason=reason,
                )
            )
        return should_process

    def _check_filter_file_extensions(self, tree_obj_path: str) -> bool:
        """
        Check if a tree object should be allowed based on the file extensions.

        :param `tree_obj_path`: path of the tree object i.e. 'llama_index/indices'

        :return: True if the tree object should be allowed, False otherwise
        """
        if self._filter_file_extensions is None:
            return True
        filter_file_extensions, filter_type = self._filter_file_extensions
        print_if_verbose(
            self._verbose,
            f"Checking {tree_obj_path} whether to {filter_type} it"
            + f" based on the filter file extensions: {filter_file_extensions}",
        )

        if filter_type == self.FilterType.EXCLUDE:
            return get_file_extension(tree_obj_path) not in filter_file_extensions
        if filter_type == self.FilterType.INCLUDE:
            return get_file_extension(tree_obj_path) in filter_file_extensions
        raise ValueError(
            f"Unknown filter type: {filter_type}. "
            "Please use either 'INCLUDE' or 'EXCLUDE'."
        )

    def _allow_tree_obj(
        self, tree_obj_path: str, tree_obj_type: str, file_size: int = None
    ) -> bool:
        """
        Check if a tree object should be allowed.

        :param `tree_obj_path`: path of the tree object

        :return: True if the tree object should be allowed, False otherwise

        """
        if self._filter_directories is not None and tree_obj_type == "tree":
            return self._check_filter_directories(tree_obj_path)

        if self._filter_file_extensions is not None and tree_obj_type == "blob":
            return self._check_filter_directories(
                tree_obj_path
            ) and self._check_filter_file_extensions(tree_obj_path)

        if self._filter_file_paths is not None and tree_obj_type == "blob":
            return self._check_filter_file_paths(
                tree_obj_path
            ) and self._check_filter_directories(tree_obj_path)

        if file_size:
            return self._check_user_callback(tree_obj_path, file_size)

        return True

    def _load_data_from_commit(self, commit_sha: str) -> List[Document]:
        """
        Load data from a commit.

        Loads github repository data from a specific commit sha.

        :param `commit`: commit sha

        :return: list of documents
        """
        repo_name = f"{self._owner}/{self._repo}"

        # Notify repository processing started
        self.dispatcher.event(
            GitHubRepositoryProcessingStartedEvent(
                repository_name=repo_name, branch_or_commit=commit_sha
            )
        )

        commit_response: GitCommitResponseModel = self._loop.run_until_complete(
            self._github_client.get_commit(
                self._owner,
                self._repo,
                commit_sha,
                timeout=self._timeout,
                retries=self._retries,
            )
        )

        tree_sha = commit_response.commit.tree.sha
        blobs_and_paths = self._loop.run_until_complete(self._recurse_tree(tree_sha))

        print_if_verbose(self._verbose, f"got {len(blobs_and_paths)} blobs")

        # Notify total files to process
        self.dispatcher.event(
            GitHubTotalFilesToProcessEvent(
                repository_name=repo_name,
                branch_or_commit=commit_sha,
                total_files=len(blobs_and_paths),
            )
        )

        documents = self._loop.run_until_complete(
            self._generate_documents(
                blobs_and_paths=blobs_and_paths,
                id=commit_sha,
            )
        )

        # Notify repository processing completed
        self.dispatcher.event(
            GitHubRepositoryProcessingCompletedEvent(
                repository_name=repo_name,
                branch_or_commit=commit_sha,
                total_documents=len(documents),
            )
        )

        return documents

    def _load_data_from_branch(
        self,
        branch: str,
        file_path: Optional[str] = None,
    ) -> List[Document]:
        """
        Load data from a branch.

        Loads github repository data from a specific branch.

        :param `branch`: branch name
        :param `file_path`: the full path to a specific file in the repo

        :return: list of documents
        """
        repo_name = f"{self._owner}/{self._repo}"

        # Notify repository processing started
        self.dispatcher.event(
            GitHubRepositoryProcessingStartedEvent(
                repository_name=repo_name, branch_or_commit=branch
            )
        )

        branch_data: GitBranchResponseModel = self._loop.run_until_complete(
            self._github_client.get_branch(
                self._owner,
                self._repo,
                branch,
                timeout=self._timeout,
                retries=self._retries,
            )
        )

        tree_sha = branch_data.commit.commit.tree.sha
        blobs_and_paths = self._loop.run_until_complete(self._recurse_tree(tree_sha))

        print_if_verbose(self._verbose, f"got {len(blobs_and_paths)} blobs")

        # Notify total files to process
        self.dispatcher.event(
            GitHubTotalFilesToProcessEvent(
                repository_name=repo_name,
                branch_or_commit=branch,
                total_files=len(blobs_and_paths),
            )
        )

        documents = self._loop.run_until_complete(
            self._generate_documents(
                blobs_and_paths=blobs_and_paths,
                id=branch,
            )
        )

        # Notify repository processing completed
        self.dispatcher.event(
            GitHubRepositoryProcessingCompletedEvent(
                repository_name=repo_name,
                branch_or_commit=branch,
                total_documents=len(documents),
            )
        )

        return documents

    def load_data(
        self,
        commit_sha: Optional[str] = None,
        branch: Optional[str] = None,
        file_path: Optional[str] = None,
    ) -> List[Document]:
        """
        Load data from a commit or a branch.

        Loads github repository data from a specific commit sha or a branch.

        :param `commit_sha`: commit sha
        :param `branch`: branch name
        :param `file_path`: the full path to a specific file in the repo

        :return: list of documents
        """
        if commit_sha is not None and branch is not None:
            raise ValueError("You can only specify one of commit or branch.")

        if commit_sha is None and branch is None:
            raise ValueError("You must specify one of commit or branch.")

        if commit_sha is not None:
            return self._load_data_from_commit(commit_sha)

        if branch is not None:
            return self._load_data_from_branch(branch, file_path=file_path)

        raise ValueError("You must specify one of commit or branch.")

    async def _recurse_tree(
        self,
        tree_sha: str,
        current_path: str = "",
        current_depth: int = 0,
        max_depth: int = -1,
    ) -> Any:
        """
        Recursively get all blob tree objects in a tree.

        And construct their full path relative to the root of the repository.
        (see GitTreeResponseModel.GitTreeObject in
            github_api_client.py for more information)

        :param `tree_sha`: sha of the tree to recurse
        :param `current_path`: current path of the tree
        :param `current_depth`: current depth of the tree
        :return: list of tuples of
            (tree object, file's full path relative to the root of the repo)
        """
        if max_depth != -1 and current_depth > max_depth:
            return []

        blobs_and_full_paths: List[Tuple[GitTreeResponseModel.GitTreeObject, str]] = []
        print_if_verbose(
            self._verbose,
            "\t" * current_depth + f"current path: {current_path}",
        )

        tree_data: GitTreeResponseModel = await self._github_client.get_tree(
            self._owner,
            self._repo,
            tree_sha,
            timeout=self._timeout,
            retries=self._retries,
        )
        print_if_verbose(
            self._verbose, "\t" * current_depth + f"tree data: {tree_data}"
        )
        print_if_verbose(
            self._verbose, "\t" * current_depth + f"processing tree {tree_sha}"
        )
        for tree_obj in tree_data.tree:
            file_path = os.path.join(current_path, tree_obj.path)
            file_size = None
            if tree_obj.type == "blob":
                file_size = tree_obj.size
            if not self._allow_tree_obj(file_path, tree_obj.type, file_size):
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"ignoring {tree_obj.path} due to filter",
                )
                continue

            print_if_verbose(
                self._verbose,
                "\t" * current_depth + f"tree object: {tree_obj}",
            )

            if tree_obj.type == "tree":
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"recursing into {tree_obj.path}",
                )

                blobs_and_full_paths.extend(
                    await self._recurse_tree(
                        tree_obj.sha, file_path, current_depth + 1, max_depth
                    )
                )
            elif tree_obj.type == "blob":
                print_if_verbose(
                    self._verbose,
                    "\t" * current_depth + f"found blob {tree_obj.path}",
                )

                blobs_and_full_paths.append((tree_obj, file_path))

            print_if_verbose(
                self._verbose,
                "\t" * current_depth + f"blob and full paths: {blobs_and_full_paths}",
            )
        return blobs_and_full_paths

    def _get_base_url(self, blob_url):
        match = re.match(r"(https://[^/]+\.com/)", blob_url)
        if match:
            return match.group(1)
        else:
            return "https://github.com/"

    async def _generate_documents(
        self,
        blobs_and_paths: List[Tuple[GitTreeResponseModel.GitTreeObject, str]],
        id: str = "",
    ) -> List[Document]:
        """
        Generate documents from a list of blobs and their full paths.

        :param `blobs_and_paths`: list of tuples of
            (tree object, file's full path in the repo relative to the root of the repo)
        :param `id`: the branch name or commit sha used when loading the repo
        :return: list of documents
        """
        buffered_iterator = BufferedGitBlobDataIterator(
            blobs_and_paths=blobs_and_paths,
            github_client=self._github_client,
            owner=self._owner,
            repo=self._repo,
            loop=self._loop,
            buffer_size=self._concurrent_requests,  # TODO: make this configurable
            verbose=self._verbose,
            timeout=self._timeout,
            retries=self._retries,
        )

        documents = []
        async for blob_data, full_path in buffered_iterator:
            print_if_verbose(self._verbose, f"generating document for {full_path}")
            assert blob_data.encoding == "base64", (
                f"blob encoding {blob_data.encoding} not supported"
            )

            # Notify file processing started
            file_extension = get_file_extension(full_path)
            self.dispatcher.event(
                GitHubFileProcessingStartedEvent(
                    file_path=full_path, file_type=file_extension
                )
            )

            decoded_bytes = None
            try:
                decoded_bytes = base64.b64decode(blob_data.content)
                del blob_data.content
            except binascii.Error:
                print_if_verbose(
                    self._verbose, f"could not decode {full_path} as base64"
                )
                self.dispatcher.event(
                    GitHubFileFailedEvent(
                        file_path=full_path,
                        file_type=file_extension,
                        error="Could not decode as base64",
                    )
                )
                continue

            try:
                if self._use_parser or file_extension in self._file_readers:
                    document = self._parse_supported_file(
                        file_path=full_path,
                        file_content=decoded_bytes,
                        tree_sha=blob_data.sha,
                        tree_path=full_path,
                    )
                    if document is not None:
                        documents.append(document)
                        self.dispatcher.event(
                            GitHubFileProcessedEvent(
                                file_path=full_path,
                                file_type=file_extension,
                                file_size=len(decoded_bytes) if decoded_bytes else 0,
                                document=document,
                            )
                        )
                        continue
                    print_if_verbose(
                        self._verbose,
                        f"could not parse {full_path} as a supported file type"
                        + " - falling back to decoding as utf-8 raw text",
                    )
            except Exception as e:
                self.logger.error(f"Error processing file {full_path}: {e}")
                self.dispatcher.event(
                    GitHubFileFailedEvent(
                        file_path=full_path,
                        file_type=get_file_extension(full_path),
                        error=str(e),
                    )
                )
                if self.fail_on_error:
                    raise
                else:
                    self.logger.warning(
                        f"Failed to process file {full_path}: {e}. Skipping this file."
                    )
                    continue

            try:
                if decoded_bytes is None:
                    raise ValueError("decoded_bytes is None")
                decoded_text = decoded_bytes.decode("utf-8")
            except UnicodeDecodeError:
                print_if_verbose(
                    self._verbose, f"could not decode {full_path} as utf-8"
                )
                self.dispatcher.event(
                    GitHubFileFailedEvent(
                        file_path=full_path,
                        file_type=file_extension,
                        error="Could not decode as UTF-8",
                    )
                )
                continue
            print_if_verbose(
                self._verbose,
                f"got {len(decoded_text)} characters"
                + f"- adding to documents - {full_path}",
            )
            url = os.path.join(
                self._get_base_url(blob_data.url),
                self._owner,
                self._repo,
                "blob/",
                id,
                full_path,
            )
            document = Document(
                text=decoded_text,
                doc_id=blob_data.sha,
                metadata={
                    "file_path": full_path,
                    "file_name": full_path.split("/")[-1],
                    "url": url,
                },
            )
            documents.append(document)
            self.dispatcher.event(
                GitHubFileProcessedEvent(
                    file_path=full_path,
                    file_type=file_extension,
                    file_size=len(decoded_text.encode("utf-8")),
                    document=document,
                )
            )

        return documents

    def _parse_supported_file(
        self,
        file_path: str,
        file_content: bytes,
        tree_sha: str,
        tree_path: str,
    ) -> Optional[Document]:
        """
        Parse a file if it is supported by a parser.

        :param `file_path`: path of the file in the repo
        :param `file_content`: content of the file
        :return: Document if the file is supported by a parser, None otherwise
        """
        file_extension = get_file_extension(file_path)
        if file_extension not in self._supported_suffix:
            # skip
            return None

        if file_extension not in self._file_readers:
            # initialize reader
            cls_ = DEFAULT_FILE_READER_CLS[file_extension]
            self._file_readers[file_extension] = cls_()

        reader = self._file_readers[file_extension]

        print_if_verbose(
            self._verbose,
            f"parsing {file_path}"
            + f"as {file_extension} with "
            + f"{reader.__class__.__name__}",
        )

        if self.custom_folder:
            with tempfile.NamedTemporaryFile(
                dir=self.custom_folder,
                suffix=f".{file_extension}",
                mode="w+b",
                delete=False,
            ) as tmpfile:
                parsed_file = self._start_parsing(
                    tmpfile, file_path, file_content, reader
                )
        else:
            with tempfile.TemporaryDirectory() as tmpdirname:
                with tempfile.NamedTemporaryFile(
                    dir=tmpdirname,
                    suffix=f".{file_extension}",
                    mode="w+b",
                    delete=False,
                ) as tmpfile:
                    parsed_file = self._start_parsing(
                        tmpfile, file_path, file_content, reader
                    )

        if parsed_file is None:
            return None
        return Document(
            text=parsed_file,
            doc_id=tree_sha,
            metadata={
                "file_path": file_path,
                "file_name": tree_path,
            },
        )

    def _start_parsing(
        self,
        tmpfile: tempfile.NamedTemporaryFile,
        file_path: str,
        file_content: bytes,
        reader: BaseReader,
    ):
        print_if_verbose(
            self._verbose,
            "created a temporary file" + f"{tmpfile.name} for parsing {file_path}",
        )
        tmpfile.write(file_content)
        tmpfile.flush()
        tmpfile.close()
        try:
            docs = reader.load_data(pathlib.Path(tmpfile.name))
            parsed_file = "\n\n".join([doc.get_text() for doc in docs])
        except Exception as e:
            print_if_verbose(self._verbose, f"error while parsing {file_path}")
            self.logger.error(
                "Error while parsing "
                + f"{file_path} with "
                + f"{reader.__class__.__name__}:\n{e}"
            )
            parsed_file = None
        finally:
            os.remove(tmpfile.name)
        return parsed_file

FilterType

Bases: Enum

Filter type.

Used to determine whether the filter is inclusive or exclusive.

Attributes:

Name	Type	Description
-	EXCLUDE	Exclude the files in the directories or with the extensions.
-	INCLUDE	Include only the files in the directories or with the extensions.

Source code in llama_index/readers/github/repository/base.py

class FilterType(enum.Enum):
    """
    Filter type.

    Used to determine whether the filter is inclusive or exclusive.

    Attributes:
        - EXCLUDE: Exclude the files in the directories or with the extensions.
        - INCLUDE: Include only the files in the directories or with the extensions.

    """

    EXCLUDE = enum.auto()
    INCLUDE = enum.auto()

load_data

load_data(commit_sha: Optional[str] = None, branch: Optional[str] = None, file_path: Optional[str] = None) -> List[Document]

Load data from a commit or a branch.

Loads github repository data from a specific commit sha or a branch.

:param commit_sha: commit sha :param branch: branch name :param file_path: the full path to a specific file in the repo

:return: list of documents

Source code in llama_index/readers/github/repository/base.py

def load_data(
    self,
    commit_sha: Optional[str] = None,
    branch: Optional[str] = None,
    file_path: Optional[str] = None,
) -> List[Document]:
    """
    Load data from a commit or a branch.

    Loads github repository data from a specific commit sha or a branch.

    :param `commit_sha`: commit sha
    :param `branch`: branch name
    :param `file_path`: the full path to a specific file in the repo

    :return: list of documents
    """
    if commit_sha is not None and branch is not None:
        raise ValueError("You can only specify one of commit or branch.")

    if commit_sha is None and branch is None:
        raise ValueError("You must specify one of commit or branch.")

    if commit_sha is not None:
        return self._load_data_from_commit(commit_sha)

    if branch is not None:
        return self._load_data_from_branch(branch, file_path=file_path)

    raise ValueError("You must specify one of commit or branch.")

GitHubAppAuth

GitHub App authentication handler.

This class manages authentication for GitHub Apps by generating JWTs and obtaining/caching installation access tokens. Tokens are automatically refreshed when they expire.

Attributes:

Name	Type	Description
app_id	str	The GitHub App ID.
private_key	str	The private key for the GitHub App (PEM format).
installation_id	str	The installation ID for the GitHub App.

Examples:

>>> # Read private key from file
>>> with open("private-key.pem", "r") as f:
...     private_key = f.read()
>>>
>>> # Create auth handler
>>> auth = GitHubAppAuth(
...     app_id="123456",
...     private_key=private_key,
...     installation_id="789012"
... )
>>>
>>> # Get installation token (cached and auto-refreshed)
>>> import asyncio
>>> token = asyncio.run(auth.get_installation_token())

Source code in llama_index/readers/github/github_app_auth.py

class GitHubAppAuth:
    """
    GitHub App authentication handler.

    This class manages authentication for GitHub Apps by generating JWTs and
    obtaining/caching installation access tokens. Tokens are automatically
    refreshed when they expire.

    Attributes:
        app_id (str): The GitHub App ID.
        private_key (str): The private key for the GitHub App (PEM format).
        installation_id (str): The installation ID for the GitHub App.

    Examples:
        >>> # Read private key from file
        >>> with open("private-key.pem", "r") as f:
        ...     private_key = f.read()
        >>>
        >>> # Create auth handler
        >>> auth = GitHubAppAuth(
        ...     app_id="123456",
        ...     private_key=private_key,
        ...     installation_id="789012"
        ... )
        >>>
        >>> # Get installation token (cached and auto-refreshed)
        >>> import asyncio
        >>> token = asyncio.run(auth.get_installation_token())

    """

    # Token expiry buffer in seconds (refresh 5 minutes before expiry)
    TOKEN_EXPIRY_BUFFER = 300
    # JWT expiry time in seconds (10 minutes, max allowed by GitHub)
    JWT_EXPIRY_SECONDS = 600
    # Installation token expiry time in seconds (1 hour, GitHub default)
    INSTALLATION_TOKEN_EXPIRY_SECONDS = 3600

    def __init__(
        self,
        app_id: str,
        private_key: str,
        installation_id: str,
        base_url: str = "https://api.github.com",
    ) -> None:
        """
        Initialize GitHubAppAuth.

        Args:
            app_id: The GitHub App ID.
            private_key: The private key for the GitHub App in PEM format.
            installation_id: The installation ID for the GitHub App.
            base_url: Base URL for GitHub API (default: "https://api.github.com").

        Raises:
            ImportError: If PyJWT is not installed.
            GitHubAppAuthenticationError: If initialization fails.

        """
        if jwt is None:
            raise ImportError(
                "PyJWT is required for GitHub App authentication. "
                "Install it with: pip install 'PyJWT[crypto]>=2.8.0'"
            )

        if not app_id:
            raise GitHubAppAuthenticationError("app_id is required")
        if not private_key:
            raise GitHubAppAuthenticationError("private_key is required")
        if not installation_id:
            raise GitHubAppAuthenticationError("installation_id is required")

        self.app_id = app_id
        self.private_key = private_key
        self.installation_id = installation_id
        self.base_url = base_url.rstrip("/")

        # Token cache
        self._token_cache: Optional[str] = None
        self._token_expires_at: float = 0

    def _generate_jwt(self) -> str:
        """
        Generate JWT for GitHub App authentication.

        The JWT is used to authenticate as the GitHub App itself, before
        obtaining an installation access token.

        Returns:
            The generated JWT token.

        Raises:
            GitHubAppAuthenticationError: If JWT generation fails.

        """
        try:
            now = int(time.time())
            payload = {
                "iat": now - 60,  # Issued at (with 60s buffer for clock skew)
                "exp": now + self.JWT_EXPIRY_SECONDS,  # Expires in 10 minutes
                "iss": self.app_id,  # Issuer is the app ID
            }

            return jwt.encode(payload, self.private_key, algorithm="RS256")
        except Exception as e:
            raise GitHubAppAuthenticationError(f"Failed to generate JWT: {e!s}") from e

    async def get_installation_token(self, force_refresh: bool = False) -> str:
        """
        Get or refresh installation access token.

        This method returns a cached token if it's still valid, or requests
        a new token from GitHub if the cached token is expired or about to expire.

        Args:
            force_refresh: If True, forces a token refresh even if cached token
                         is still valid.

        Returns:
            A valid installation access token.

        Raises:
            GitHubAppAuthenticationError: If token retrieval fails.
            ImportError: If httpx is not installed.

        """
        # Check if cached token is still valid (with buffer)
        if not force_refresh and self._is_token_valid():
            return self._token_cache  # type: ignore

        # Generate new token
        try:
            import httpx
        except ImportError:
            raise ImportError(
                "httpx is required for GitHub App authentication. "
                "Install it with: pip install httpx>=0.26.0"
            )

        jwt_token = self._generate_jwt()

        headers = {
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {jwt_token}",
            "X-GitHub-Api-Version": "2022-11-28",
        }

        url = f"{self.base_url}/app/installations/{self.installation_id}/access_tokens"

        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(url, headers=headers, timeout=10.0)
                response.raise_for_status()
                data = response.json()

                self._token_cache = data["token"]
                # Token typically expires in 1 hour
                self._token_expires_at = (
                    time.time() + self.INSTALLATION_TOKEN_EXPIRY_SECONDS
                )

                return self._token_cache
        except httpx.HTTPStatusError as e:
            raise GitHubAppAuthenticationError(
                f"Failed to get installation token: {e.response.status_code} - {e.response.text}"
            ) from e
        except Exception as e:
            raise GitHubAppAuthenticationError(
                f"Failed to get installation token: {e!s}"
            ) from e

    def _is_token_valid(self) -> bool:
        """
        Check if the cached token is still valid.

        Returns:
            True if token exists and is not expired (accounting for buffer).

        """
        if not self._token_cache:
            return False

        # Check if token will expire within the buffer period
        time_until_expiry = self._token_expires_at - time.time()
        return time_until_expiry > self.TOKEN_EXPIRY_BUFFER

    def invalidate_token(self) -> None:
        """
        Invalidate the cached token.

        This forces the next call to get_installation_token() to fetch a new token.
        Useful if you know the token has been revoked or is no longer valid.

        """
        self._token_cache = None
        self._token_expires_at = 0

get_installation_token async

get_installation_token(force_refresh: bool = False) -> str

Get or refresh installation access token.

This method returns a cached token if it's still valid, or requests a new token from GitHub if the cached token is expired or about to expire.

Parameters:

Name	Type	Description	Default
force_refresh	bool	If True, forces a token refresh even if cached token is still valid.	False

Returns:

Type	Description
str	A valid installation access token.

Raises:

Type	Description
GitHubAppAuthenticationError	If token retrieval fails.
ImportError	If httpx is not installed.

Source code in llama_index/readers/github/github_app_auth.py

async def get_installation_token(self, force_refresh: bool = False) -> str:
    """
    Get or refresh installation access token.

    This method returns a cached token if it's still valid, or requests
    a new token from GitHub if the cached token is expired or about to expire.

    Args:
        force_refresh: If True, forces a token refresh even if cached token
                     is still valid.

    Returns:
        A valid installation access token.

    Raises:
        GitHubAppAuthenticationError: If token retrieval fails.
        ImportError: If httpx is not installed.

    """
    # Check if cached token is still valid (with buffer)
    if not force_refresh and self._is_token_valid():
        return self._token_cache  # type: ignore

    # Generate new token
    try:
        import httpx
    except ImportError:
        raise ImportError(
            "httpx is required for GitHub App authentication. "
            "Install it with: pip install httpx>=0.26.0"
        )

    jwt_token = self._generate_jwt()

    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {jwt_token}",
        "X-GitHub-Api-Version": "2022-11-28",
    }

    url = f"{self.base_url}/app/installations/{self.installation_id}/access_tokens"

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(url, headers=headers, timeout=10.0)
            response.raise_for_status()
            data = response.json()

            self._token_cache = data["token"]
            # Token typically expires in 1 hour
            self._token_expires_at = (
                time.time() + self.INSTALLATION_TOKEN_EXPIRY_SECONDS
            )

            return self._token_cache
    except httpx.HTTPStatusError as e:
        raise GitHubAppAuthenticationError(
            f"Failed to get installation token: {e.response.status_code} - {e.response.text}"
        ) from e
    except Exception as e:
        raise GitHubAppAuthenticationError(
            f"Failed to get installation token: {e!s}"
        ) from e

invalidate_token

invalidate_token() -> None

Invalidate the cached token.

This forces the next call to get_installation_token() to fetch a new token. Useful if you know the token has been revoked or is no longer valid.

Source code in llama_index/readers/github/github_app_auth.py

def invalidate_token(self) -> None:
    """
    Invalidate the cached token.

    This forces the next call to get_installation_token() to fetch a new token.
    Useful if you know the token has been revoked or is no longer valid.

    """
    self._token_cache = None
    self._token_expires_at = 0

GitHubAppAuthenticationError

Bases: Exception

Raised when GitHub App authentication fails.

Source code in llama_index/readers/github/github_app_auth.py

class GitHubAppAuthenticationError(Exception):
    """Raised when GitHub App authentication fails."""

options: members: - GitHubRepositoryCollaboratorsReader - GitHubRepositoryIssuesReader - GithubRepositoryReader