tessl/pypi-gidgethub

An asynchronous GitHub API library designed as a sans-I/O library for GitHub API access

—

Pending

Overview

Eval results

Files

API Client

Name: tessl/pypi-gidgethub
Author: tessl

High-level GitHub API interface providing HTTP method implementations (GET, POST, PUT, PATCH, DELETE) with authentication, caching, and GraphQL support. The abstract base class defines the interface that concrete HTTP implementations must fulfill.

Capabilities

Abstract Base Class

The core abstract class that defines the GitHub API interface.

import abc
from typing import Any, AsyncGenerator, Dict, Mapping, MutableMapping, Optional, Tuple
from uritemplate import variable
from gidgethub.sansio import RateLimit
import gidgethub.sansio as sansio

# Cache type definition
CACHE_TYPE = MutableMapping[str, Tuple[Optional[str], Optional[str], Any, Optional[str]]]

class GitHubAPI(abc.ABC):
    """Provide an idiomatic API for making calls to GitHub's API."""
    
    def __init__(
        self,
        requester: str,
        *,
        oauth_token: Optional[str] = None,
        cache: Optional[CACHE_TYPE] = None,
        base_url: str = "https://api.github.com"
    ) -> None:
        """
        Initialize GitHub API client.
        
        Parameters:
        - requester: User agent identifier (username or project name)
        - oauth_token: Personal access token for authentication
        - cache: Optional cache for GET requests (etag/last-modified based)
        - base_url: GitHub API base URL
        """
    
    # Abstract methods that implementations must provide
    @abc.abstractmethod
    async def _request(
        self, 
        method: str, 
        url: str, 
        headers: Mapping[str, str], 
        body: bytes = b""
    ) -> Tuple[int, Mapping[str, str], bytes]:
        """Make an HTTP request."""
    
    @abc.abstractmethod
    async def sleep(self, seconds: float) -> None:
        """Sleep for the specified number of seconds."""
    
    # Attributes
    requester: str
    oauth_token: Optional[str]
    rate_limit: Optional[RateLimit]
    base_url: str

GET Methods

Methods for retrieving data from the GitHub API.

async def getitem(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Any:
    """
    Send a GET request for a single item to the specified endpoint.
    
    Parameters:
    - url: API endpoint URL (absolute or relative)
    - url_vars: Variables for URI template expansion
    - accept: Accept header value for API version/format
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    
    Returns:
    - Parsed JSON response data
    """

async def getstatus(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
) -> int:
    """
    Send a GET request and return only the status code.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    
    Returns:
    - HTTP status code (even for error responses)
    """

async def getiter(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
    iterable_key: Optional[str] = "items",
) -> AsyncGenerator[Any, None]:
    """
    Return an async iterable for all items at a specified endpoint.
    Automatically handles pagination using Link headers.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    - iterable_key: Key containing items in paginated responses
    
    Yields:
    - Individual items from paginated API responses
    """

Mutation Methods

Methods for creating, updating, and deleting data via the GitHub API.

async def post(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    data: Any,
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
    content_type: str = "application/json",
) -> Any:
    """
    Send a POST request to create resources.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - data: Request body data (JSON serializable or bytes)
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    - content_type: Content-Type header value
    
    Returns:
    - Parsed JSON response data
    """

async def patch(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    data: Any,
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Any:
    """
    Send a PATCH request to update resources.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - data: Request body data (JSON serializable)
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    
    Returns:
    - Parsed JSON response data
    """

async def put(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    data: Any = b"",
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Any:
    """
    Send a PUT request to create or replace resources.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - data: Request body data (default: empty)
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    
    Returns:
    - Parsed JSON response data
    """

async def delete(
    self,
    url: str,
    url_vars: Optional[variable.VariableValueDict] = {},
    *,
    data: Any = b"",
    accept: str = sansio.accept_format(),
    jwt: Optional[str] = None,
    oauth_token: Optional[str] = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> None:
    """
    Send a DELETE request to remove resources.
    
    Parameters:
    - url: API endpoint URL
    - url_vars: Variables for URI template expansion
    - data: Request body data (default: empty)
    - accept: Accept header value
    - jwt: JWT token for GitHub App authentication
    - oauth_token: OAuth token (overrides instance token)
    - extra_headers: Additional headers to include
    
    Returns:
    - None (DELETE responses typically have no body)
    """

GraphQL Support

Methods for querying GitHub's GraphQL v4 API.

async def graphql(
    self,
    query: str,
    *,
    endpoint: str = "https://api.github.com/graphql",
    **variables: Any,
) -> Any:
    """
    Query the GraphQL v4 API.
    
    Parameters:
    - query: GraphQL query string
    - endpoint: GraphQL endpoint URL
    - **variables: Query variables as keyword arguments
    
    Returns:
    - GraphQL response data
    
    Raises:
    - GraphQLException: For malformed responses
    - GraphQLAuthorizationFailure: For 401 responses
    - BadGraphQLRequest: For 4XX responses
    - QueryError: For GraphQL query errors
    - GitHubBroken: For 5XX responses
    """

Usage Examples

Basic Repository Information

import asyncio
from gidgethub.aiohttp import GitHubAPI
import aiohttp

async def get_repo_info():
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "my-app/1.0")
        
        # Get single repository
        repo = await gh.getitem("/repos/octocat/Hello-World")
        print(f"Repository: {repo['name']}")
        print(f"Stars: {repo['stargazers_count']}")
        
        # Check if repository exists (status only)
        status = await gh.getstatus("/repos/octocat/Hello-World")
        if status == 200:
            print("Repository exists")

asyncio.run(get_repo_info())

Pagination with Async Iteration

async def list_all_issues():
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "my-app/1.0", oauth_token="your_token")
        
        # Iterate through all issues across all pages
        async for issue in gh.getiter("/repos/owner/repo/issues"):
            print(f"Issue #{issue['number']}: {issue['title']}")

Creating and Updating Resources

async def manage_issues():
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "my-app/1.0", oauth_token="your_token")
        
        # Create a new issue
        new_issue = await gh.post(
            "/repos/owner/repo/issues",
            data={
                "title": "Bug report",
                "body": "Description of the bug",
                "labels": ["bug", "urgent"]
            }
        )
        issue_number = new_issue['number']
        
        # Update the issue
        updated_issue = await gh.patch(
            f"/repos/owner/repo/issues/{issue_number}",
            data={"state": "closed"}
        )
        
        # Add a comment
        await gh.post(
            f"/repos/owner/repo/issues/{issue_number}/comments",
            data={"body": "Fixed in latest release"}
        )

GraphQL Queries

async def graphql_example():
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "my-app/1.0", oauth_token="your_token")
        
        query = """
        query($owner: String!, $name: String!) {
            repository(owner: $owner, name: $name) {
                name
                description
                stargazerCount
                forkCount
                issues(first: 10) {
                    nodes {
                        title
                        state
                    }
                }
            }
        }
        """
        
        result = await gh.graphql(query, owner="octocat", name="Hello-World")
        repo = result['repository']
        print(f"Repository: {repo['name']}")
        print(f"Issues: {len(repo['issues']['nodes'])}")

URL Template Variables

async def template_variables():
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "my-app/1.0")
        
        # Use template variables for dynamic URLs
        repo_data = await gh.getitem(
            "/repos/{owner}/{repo}",
            {"owner": "octocat", "repo": "Hello-World"}
        )
        
        # Get specific issue
        issue = await gh.getitem(
            "/repos/{owner}/{repo}/issues/{issue_number}",
            {
                "owner": "octocat", 
                "repo": "Hello-World",
                "issue_number": 1
            }
        )

Constants

JSON_CONTENT_TYPE: str = "application/json"
UTF_8_CHARSET: str = "utf-8" 
JSON_UTF_8_CHARSET: str = "application/json; charset=utf-8"
ITERABLE_KEY: str = "items"

Types

from typing import Any, AsyncGenerator, Dict, Mapping, MutableMapping, Optional, Tuple
from uritemplate import variable

# Cache type for HTTP caching
CACHE_TYPE = MutableMapping[str, Tuple[Optional[str], Optional[str], Any, Optional[str]]]

# From sansio module
RateLimit = gidgethub.sansio.RateLimit

Install with Tessl CLI