GitHub

GitHub仓库协作者读取器 #

基类：EventBaseReader

GitHub 仓库协作者读取器。

获取 GitHub 仓库的协作者列表并返回文档列表。

示例：

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

workflows/handler.py 中的源代码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

过滤器类型 #

基类：EventEnum

筛选类型。

用于确定过滤器是包含性还是排除性的。

workflows/handler.py 中的源代码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[文档]

GitHub 仓库协作者读取器。

获取 GitHub 仓库中的协作者列表并将其转换为文档。

每个协作者通过以下步骤被转换为文档：

- 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: 文档列表

workflows/handler.py 中的源代码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

GitHub协作客户端 #

一个用于与GitHub协作者API交互的异步客户端。

客户端支持两种认证方式： 1. 个人访问令牌 (PAT) - 通过 github_token 参数或 GITHUB_TOKEN 环境变量传递 2. GitHub 应用 - 通过 github_app_auth 参数传递

示例：

>>> # 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)

workflows/handler.py 中的源代码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]

获取所有可用的端点。

workflows/handler.py 中的源代码llama_index/readers/github/collaborators/github_client.py

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

请求 `async` #

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

向 GitHub API 发起 API 请求。

参数：

名称	描述	默认
- `endpoint (str)`	要发送请求的端点名称。	required
- `method (str)`	用于请求的HTTP方法。	required
- `headers (dict)`	请求中包含的HTTP头部信息。	required
- `**kwargs`	传递给端点 URL 的关键字参数。	required

返回：

类型	描述
`Any`	`response (httpx.Response)`: API请求的响应。

引发：

类型	描述
`-ImportError`	如果未安装 `httpx` 库。
`-HTTPError`	如果API请求失败。

示例：

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

workflows/handler.py 中的源代码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

列出仓库中的协作者。

参数：

名称	类型	描述	默认
- `owner (str)`		仓库所有者。	required
- `repo (str)`		仓库名称。	required

返回：

类型	描述
`Dict`	See https://docs.github.com/en/rest/collaborators/collaborators?apiVersion=2022-11-28#list-repository-collaborators

示例：

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

workflows/handler.py 中的源代码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()

GitHub问题客户端 #

一个用于与GitHub问题API交互的异步客户端。

客户端支持两种认证方式： 1. 个人访问令牌 (PAT) - 通过 github_token 参数或 GITHUB_TOKEN 环境变量传递 2. GitHub 应用 - 通过 github_app_auth 参数传递

示例：

>>> # 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)

workflows/handler.py 中的源代码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]

获取所有可用的端点。

workflows/handler.py 中的源代码llama_index/readers/github/issues/github_client.py

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

请求 `async` #

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

向 GitHub API 发起 API 请求。

参数：

名称	描述	默认
- `endpoint (str)`	要向其发出请求的端点名称。	required
- `method (str)`	用于请求的HTTP方法。	required
- `headers (dict)`	请求中包含的HTTP头部信息。	required
- `**kwargs`	传递给端点 URL 的关键字参数。	required

返回：

类型	描述
`Any`	`response (httpx.Response)`: API请求的响应结果。

引发：

类型	描述
`-ImportError`	如果未安装 `httpx` 库。
`-HTTPError`	如果API请求失败。

示例：

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

workflows/handler.py 中的源代码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

列出仓库中的问题。

注意：GitHub的REST API将每个拉取请求都视为一个问题，但并非每个问题都是拉取请求。因此，“问题”接口可能在响应中同时返回问题和拉取请求。您可以通过pull_request键来识别拉取请求。请注意，从“问题”接口返回的拉取请求ID将是一个问题ID。要查找拉取请求ID，请使用“列出拉取请求”接口。

参数：

名称	描述	默认
- `owner (str)`	仓库所有者。	required
- `repo (str)`	仓库名称。	required
- `state (str)`	表示要返回的问题状态。默认值：open 可选值：open, closed, all。	required

返回：

类型	描述
`Dict`	See https://docs.github.com/en/rest/issues/issues?apiVersion=2022-11-28#list-repository-issues

示例：

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

workflows/handler.py 中的源代码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()

GitHub仓库问题读取器 #

基类：EventBaseReader

GitHub 仓库问题读取器。

获取 GitHub 仓库的问题列表并返回文档列表。

示例：

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

workflows/handler.py 中的源代码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

问题状态 #

基类：EventEnum

问题类型。

用于决定检索哪些问题。

属性：

名称	类型	描述
`-`	`OPEN`	只需开启问题。这是默认设置。
`-`	`CLOSED`	刚刚关闭了问题。
`-`	`ALL`	所有问题，包括已打开和已关闭的。

workflows/handler.py 中的源代码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"

过滤器类型 #

基类：EventEnum

筛选类型。

用于确定过滤器是包含性还是排除性的。

workflows/handler.py 中的源代码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[文档]

从代码仓库加载问题并将其转换为文档。

每个问题通过以下步骤转换为文档：

文档文本由问题标题和正文内容拼接而成。
文档标题即为议题标题。
文档的 doc_id 即为问题编号。
文档的 extra_info 是一个包含以下键的字典：
- state: 议题的状态。
- created_at: 议题创建的日期。
- closed_at: 议题关闭的日期。仅当议题已关闭时存在。
- url: 议题的URL地址。
- assignee: 被分配处理该议题的用户的登录名。仅当议题已分配时存在。
文档的嵌入为 None。
文档的 doc_hash 为 None。

参数：

名称	类型	描述	默认
`- state`	`IssueState`	要检索的问题状态。默认为 IssueState.OPEN。	required
`- labelFilters`		一个可选的筛选器列表，用于根据标签对问题列表进行筛选。	required

:return: 文档列表

workflows/handler.py 中的源代码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

Github客户端 #

一个用于与Github API交互的异步客户端。

此客户端用于向Github发起API请求。它提供了访问Github API端点的方法。该客户端支持两种认证方式： 1. 个人访问令牌（PAT）- 通过github_token参数或GITHUB_TOKEN环境变量传递 2. GitHub应用 - 通过github_app_auth参数传递

示例：

>>> # 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)

workflows/handler.py 中的源代码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]

获取所有可用的端点。

workflows/handler.py 中的源代码llama_index/readers/github/repository/github_client.py

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

请求 `async` #

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

向 Github API 发起 API 请求。

此方法用于向 Github API 发起 API 请求。它在客户端内部被其他方法调用。

参数：

名称	描述	默认
- `endpoint (str)`	要向其发出请求的端点名称。	required
- `method (str)`	用于请求的HTTP方法。	required
- `headers (dict)`	请求中包含的HTTP头部信息。	required
- `timeout (int or None)`	请求超时时间，单位为秒。默认值为5。	required
- `retries (int)`	请求的重试次数。默认值为0。	required
- `**kwargs`	传递给端点 URL 的关键字参数。	required

返回：

类型	描述
`Any`	`response (httpx.Response)`response (httpx.Response): API请求的响应结果。

引发：

类型	描述
`-ImportError`	如果未安装 `httpx` 库。
`-HTTPError`	如果API请求失败且 fail_on_http_error 为 True。

示例：

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

workflows/handler.py 中的源代码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

获取分支信息。（Github API 接口：getBranch）。

参数：

名称	描述	默认
- `owner (str)`	仓库所有者。	required
- `repo (str)`	仓库名称。	required
- `branch (str)`	分支名称。	required
- `branch_name (str)`	分支名称（替代 `branch`）。	required
- `timeout (int or None)`	请求超时时间，单位为秒。默认值为5。	required
- `retries (int)`	请求的重试次数。默认值为0。	required

返回：

类型	描述
`GitBranchResponseModel`	`branch_info (GitBranchResponseModel)`: 关于分支的信息。

示例：

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

workflows/handler.py 中的源代码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

获取关于树的信息。（GitHub API 端点：getTree）。

参数：

名称	描述	默认
- `owner (str)`	仓库所有者。	required
- `repo (str)`	仓库名称。	required
- `tree_sha (str)`	树的 SHA 值。	required
- `timeout (int or None)`	请求超时时间，单位为秒。默认值为5。	required
- `retries (int)`	请求的重试次数。默认值为0。	required

返回：

类型	描述
`GitTreeResponseModel`	`tree_info (GitTreeResponseModel)`: 关于树的信息。

示例：

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

workflows/handler.py 中的源代码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]

获取关于blob的信息。(Github API接口：getBlob)。

参数：

名称	描述	默认
- `owner (str)`	仓库所有者。	required
- `repo (str)`	仓库名称。	required
- `file_sha (str)`	文件的SHA值。	required
- `timeout (int or None)`	请求超时时间，单位为秒。默认值为5。	required
- `retries (int)`	请求的重试次数。默认值为0。	required

返回：

类型	描述
`Optional[GitBlobResponseModel]`	`blob_info (GitBlobResponseModel)`: 关于该二进制大对象的信息。

示例：

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

workflows/handler.py 中的源代码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

获取关于某次提交的信息。（Github API 接口：getCommit）。

参数：

名称	描述	默认
- `owner (str)`	仓库所有者。	required
- `repo (str)`	仓库名称。	required
- `commit_sha (str)`	提交的SHA值。	required
- `timeout (int or None)`	请求超时时间，单位为秒。默认值为5。	required
- `retries (int)`	请求的重试次数。默认值为0。	required

返回：

类型	描述
`GitCommitResponseModel`	`commit_info (GitCommitResponseModel)`: 关于提交的信息。

示例：

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

workflows/handler.py 中的源代码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
    )

GitHub仓库读取器 #

基类：EventBaseReader

GitHub 仓库读取器。

获取Github仓库的内容并返回文档列表。这些文档要么是仓库中文件的内容，要么是使用解析器从文件中提取的文本。

示例：

>>> 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")

workflows/handler.py 中的源代码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

过滤器类型 #

基类：EventEnum

筛选类型。

用于确定过滤器是包含性还是排除性的。

属性：

名称	类型	描述
`-`	`EXCLUDE`	排除指定目录中或具有指定扩展名的文件。
`-`	`INCLUDE`	仅包含指定目录中或具有指定扩展名的文件。

workflows/handler.py 中的源代码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[文档]

从提交或分支加载数据。

从特定的提交 SHA 或分支加载 GitHub 仓库数据。

:param commit_sha: 提交哈希值 :param branch: 分支名称 :param file_path: 仓库中特定文件的完整路径

:return: 文档列表

workflows/handler.py 中的源代码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.")

GitHub应用认证 #

GitHub 应用认证处理器。

该类通过生成JWT并获取/缓存安装访问令牌来管理GitHub应用程序的身份验证。令牌在过期时会自动刷新。

属性：

名称	类型	描述
`app_id`	`str`	GitHub 应用 ID。
`private_key`	`str`	GitHub 应用的私钥（PEM 格式）。
`installation_id`	`str`	GitHub 应用的安装ID。

示例：

>>> # 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())

workflows/handler.py 中的源代码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

获取或刷新安装访问令牌。

如果缓存的令牌仍然有效，此方法将返回缓存的令牌；如果缓存的令牌已过期或即将过期，则向GitHub请求新令牌。

参数：

名称	类型	描述	默认
`force_refresh`	`bool`	如果为True，即使缓存的令牌仍然有效，也会强制刷新令牌。	`False`

返回：

类型	描述
`str`	一个有效的安装访问令牌。

引发：

类型	描述
`GitHubAppAuthenticationError`	如果令牌检索失败。
`ImportError`	如果未安装httpx。

workflows/handler.py 中的源代码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

使缓存的令牌失效。

这会强制下一次调用 get_installation_token() 时获取新的令牌。如果您知道令牌已被撤销或不再有效，这将非常有用。

workflows/handler.py 中的源代码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

GitHub应用认证错误 #

基类：EventException

当 GitHub 应用认证失败时触发。

workflows/handler.py 中的源代码llama_index/readers/github/github_app_auth.py

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

选项：成员：- GitHubRepositoryCollaboratorsReader - GitHubRepositoryIssuesReader - GithubRepositoryReader

GitHub

GitHub仓库协作者读取器 #

过滤器类型 #

load_data #

GitHub协作客户端 #

get_all_endpoints #

请求 async #

get_collaborators async #

GitHub问题客户端 #

get_all_endpoints #

请求 async #

get_issues async #

GitHub仓库问题读取器 #

问题状态 #

过滤器类型 #

load_data #

Github客户端 #

get_all_endpoints #

请求 async #

get_branch async #

get_tree async #

get_blob async #

get_commit async #

GitHub仓库读取器 #

过滤器类型 #

load_data #

GitHub应用认证 #

get_installation_token async #

invalidate_token #

GitHub应用认证错误 #

请求 `async` #

get_collaborators `async` #

请求 `async` #

get_issues `async` #

请求 `async` #

get_branch `async` #

get_tree `async` #

get_blob `async` #

get_commit `async` #

get_installation_token `async` #