GitHubWrapper/github/internals/http.py

from __future__ import annotations

__all__ = ("HTTPClient",)

import asyncio
import logging
import platform
import time
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any, Awaitable, Dict, List, Literal, NamedTuple, Optional, Union

from aiohttp import ClientSession, __version__ as aiohttp_version

from ..errors import error_from_request
from ..utils import human_readable_time_until

try:
    import orjson  # type: ignore
except ImportError:
    import json

    json_loads = json.loads
else:
    json_loads = orjson.loads

if TYPE_CHECKING:
    from aiohttp import BasicAuth
    from typing_extensions import Self

    from ..objects import File
    from ..types import Author, Committer, OptionalAuthor, OptionalCommitter, SecurityAndAnalysis


log = logging.getLogger("github")


class RateLimits(NamedTuple):
    remaining: int
    used: int
    total: int
    reset_time: datetime
    last_request: datetime


# ====== STYLE GUIDE ===== #
# All route method names should be
# The exact same from the GitHub API
# Docs, excluding 'the', 'a' etc
# Names should be shortened, e.g:
# Information -> Info
# Repository -> Repo

# ========= TODO ========= #
# Make a good paginator
# Make objects for all API Types
# Make the requests return TypedDicts with those objects
# Make specific errors
# Make it so an error gets raised when the cooldown is reached
# Make markdown raw request route (???)

# === ROUTES CHECKLIST === #
# Actions
# Activity
# Apps
# Billing
# Branches
# Checks
# Codes of conduct           DONE
# Code Scanning
# Codespaces
# Collaborators
# Commits
# Dependabot
# Dependency Graph
# Deploy keys                DONE
# Deployments
# Emojis                     DONE
# Enterprise administration
# Gists                      DONE
# Git database
# Gitignore                  DONE
# Interactions
# Issues
# Licenses                   DONE
# Markdown                   DONE
# Meta                       DONE
# Metrics
# Migrations
# OAuth authorizations
# Organizations
# Packages
# Pages
# Projects
# Pulls
# Rate limit
# Reactions
# Releases
# Repositories               DONE
# SCIM
# Search                     DONE
# Teams
# Users                      DONE
# Webhooks


class HTTPClient:
    __session: ClientSession
    _rates: RateLimits
    _last_ping: float
    _latency: float

    def __new__(cls, **kwargs: Any) -> Awaitable[HTTPClient]:
        # Basically async def __init__
        return cls.__async_init(**kwargs)

    @classmethod
    async def __async_init(
        cls,
        *,
        headers: Optional[Dict[str, str]] = None,
        auth: Optional[BasicAuth] = None,
    ) -> HTTPClient:
        self = super(cls, cls).__new__(cls)

        headers = headers or {}

        headers.setdefault(
            "User-Agent",
            "GitHub-API-Wrapper (https://github.com/Varmonke/GitHub-API-Wrapper) @"
            f" 2.0.0a CPython/{platform.python_version()} aiohttp/{aiohttp_version}",
        )

        self.__session = ClientSession(headers=headers, auth=auth)

        time_0 = datetime.fromtimestamp(0)

        self._rates = RateLimits(60, 0, 60, time_0, time_0)

        self._last_ping = float("-inf")
        self._latency = 0

        return self

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, *_) -> None:
        await self.__session.close()

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__}>"

    @property
    def is_ratelimited(self) -> bool:
        return self._rates.remaining < 2

    async def latency(self) -> float:
        last_ping = self._last_ping

        # If there was no ping or the last ping was more than 5 seconds ago (or is currently ratelimited).
        if not self.is_ratelimited and time.monotonic() > last_ping + 5:
            self._last_ping = time.monotonic()

            start = time.monotonic()
            await self.get_github_api_root()
            self._latency = time.monotonic() - start

        return self._latency

    async def request(
        self, method: Literal["GET", "POST", "PUT", "DELETE", "PATCH"], path: str, /, **kwargs: Any
    ):
        if self.is_ratelimited:
            log.info(
                "Ratelimit exceeded, trying again in"
                f" {human_readable_time_until(self._rates.reset_time - datetime.now(timezone.utc))} (URL:"
                f" {path}, method: {method})"
            )

            # TODO: I get about 3-4 hours of cooldown
            # this might not be a good idea, might make
            # this raise an error instead.
            await asyncio.sleep(
                max((self._rates.reset_time - datetime.now(timezone.utc)).total_seconds(), 0)
            )

        async with self.__session.request(
            method, f"https://api.github.com{path}", **kwargs
        ) as response:
            headers = response.headers

            self._rates = RateLimits(
                int(headers["X-RateLimit-Remaining"]),
                int(headers["X-RateLimit-Used"]),
                int(headers["X-RateLimit-Limit"]),
                datetime.fromtimestamp(int(headers["X-RateLimit-Reset"])).replace(
                    tzinfo=timezone.utc
                ),
                datetime.now(timezone.utc),
            )

            if 200 <= response.status <= 299:
                data = await response.text(encoding="utf-8")

                if response.headers["Content-Type"] == "application/json":
                    return json_loads(data)

                return data

            raise error_from_request(response)

    # === ROUTES === #

    # === USERS === #

    async def get_authenticated_user(self):
        return await self.request("GET", "/user")

    async def update_authenticated_user(
        self,
        *,
        name: Optional[str] = None,
        email: Optional[str] = None,
        blog: Optional[str] = None,
        twitter_username: Optional[str] = None,
        company: Optional[str] = None,
        location: Optional[str] = None,
        hireable: Optional[str] = None,
        bio: Optional[str] = None,
    ):
        data = {}

        if name:
            data["name"] = name
        if email:
            data["email"] = email
        if blog:
            data["blog"] = blog
        if twitter_username:
            data["twitter_username"] = twitter_username
        if company:
            data["company"] = company
        if location:
            data["location"] = location
        if hireable:
            data["hireable"] = hireable
        if bio:
            data["bio"] = bio

        return await self.request("PATCH", "/user", json=data)

    async def list_users(self, *, since: Optional[int] = None, per_page: Optional[int] = None):
        params = {}

        if since:
            params["since"] = since
        if per_page:
            params["per_page"] = per_page

        return await self.request("GET", "/users", params=params)

    async def get_user(self, *, username: str):
        return await self.request("GET", f"/users/{username}")

    async def get_context_info_for_user(
        self,
        *,
        username: str,
        subject_type: Optional[
            Literal["organization", "repository", "user", "pull_request"]
        ] = None,
        subject_id: Optional[int] = None,
    ):
        params = {}

        if subject_type:
            params["subject_type"] = subject_type
        if subject_id:
            params["subject_id"] = subject_id

        return await self.request("GET", f"/users/{username}/hovercard", params=params)

    async def list_blocked_users_for_authenticated_user(self):
        return await self.request("GET", "/user/blocks")

    async def check_user_blocked_for_authenticated_user(self, *, username: str):
        return await self.request("GET", f"/user/blocks/{username}")

    async def block_user(self, *, username: str):
        return await self.request("PUT", f"/user/blocks/{username}")

    async def unblock_user(self, *, username: str):
        return await self.request("DELETE", f"/user/blocks/{username}")

    async def set_primary_email_visibility_for_authenticated_user(
        self, *, visibility: Literal["public", "private"]
    ):
        return await self.request(
            "PATCH", "/user/email/visibility", json={"visibility": visibility}
        )

    async def list_email_addresses_for_the_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/emails", params=params)

    async def add_email_addresses_for_the_authenticated_user(self, *, emails: List[str]):
        return await self.request("POST", "/user/emails", json={"email": emails})

    async def delete_email_addresses_for_the_authenticated_user(self, *, emails: List[str]):
        return await self.request("DELETE", "/user/emails", json={"email": emails})

    async def list_public_email_addresses_for_the_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/emails/public", params=params)

    async def list_followers_of_the_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/followers", params=params)

    async def list_following_for_the_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/following", params=params)

    async def check_person_followed_by_authenticated_user(self, *, username: str):
        return await self.request("GET", f"/user/following/{username}")

    async def follow_user(self, *, username: str):
        return await self.request("PUT", f"/user/following/{username}")

    async def unfollow_user(self, *, username: str):
        return await self.request("DELETE", f"/user/following/{username}")

    async def list_followers_for_user(
        self, *, username: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/followers", params=params)

    async def list_following_for_user(
        self, *, username: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/following", params=params)

    async def check_user_follows_another_user(self, *, username: str, target_user: str):
        return await self.request("GET", f"/users/{username}/following/{target_user}")

    async def list_gpg_keys_for_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/gpg_keys", params=params)

    async def create_gpg_key_for_authenticated_user(
        self, *, name: Optional[str] = None, armored_public_key: str
    ):
        data = {
            "armored_public_key": armored_public_key,
        }

        if name:
            data["name"] = name

        return await self.request("POST", "/user/gpg_keys", json=data)

    async def get_gpg_key_for_authenticated_user(self, *, gpg_key_id: int):
        return await self.request("GET", f"/user/gpg_keys/{gpg_key_id}")

    async def delete_gpg_key_for_authenticated_user(self, *, gpg_key_id: int):
        return await self.request("DELETE", f"/user/gpg_keys/{gpg_key_id}")

    async def list_gpg_keys_for_user(
        self, *, username: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/gpg_keys", params=params)

    async def list_public_ssh_keys_for_authenticated_user(
        self, *, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/user/keys", params=params)

    async def create_public_ssh_key_for_authenticated_user(
        self, *, title: Optional[str] = None, key: str
    ):
        data = {
            "key": key,
        }

        if title:
            data["title"] = title

        return await self.request("POST", "/user/keys", json=data)

    async def get_public_ssh_key_for_authenticated_user(self, *, key_id: int):
        return await self.request("GET", f"/user/keys/{key_id}")

    async def delete_public_ssh_key_for_authenticated_user(self, *, key_id: int):
        return await self.request("DELETE", f"/user/keys/{key_id}")

    async def list_public_ssh_keys_for_user(
        self, *, username: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/keys", params=params)

    # === REPOS === #

    async def list_org_repos(
        self,
        *,
        org: str,
        type: Optional[
            Literal["all", "public", "private", "forks", "sources", "member", "internal"]
        ] = None,
        sort: Optional[Literal["created", "updated", "pushed", "full_name"]] = None,
        direction: Optional[Literal["asc", "desc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if type:
            params["type"] = type
        if sort:
            params["sort"] = sort
        if direction:
            params["direction"] = direction
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/orgs/{org}/repos", params=params)

    async def create_org_repo(
        self,
        *,
        org: str,
        name: str,
        description: Optional[str] = None,
        homepage: Optional[str] = None,
        private: Optional[bool] = None,
        visibility: Optional[Literal["public", "private", "internal"]] = None,
        has_issues: Optional[bool] = None,
        has_projects: Optional[bool] = None,
        has_wiki: Optional[bool] = None,
        is_template: Optional[bool] = None,
        team_id: Optional[int] = None,
        auto_init: Optional[bool] = None,
        gitignore_template: Optional[str] = None,
        license_template: Optional[str] = None,
        allow_squash_merge: Optional[bool] = None,
        allow_merge_commit: Optional[bool] = None,
        allow_rebase_merge: Optional[bool] = None,
        allow_auto_merge: Optional[bool] = None,
        delete_branch_on_merge: Optional[bool] = None,
        use_squash_pr_title_as_default: Optional[bool] = None,
    ):
        data: Dict[str, Union[str, bool, int]] = {
            "name": name,
        }

        if description:
            data["description"] = description
        if homepage:
            data["homepage"] = homepage
        if private:
            data["private"] = private
        if visibility:
            data["visibility"] = visibility
        if has_issues:
            data["has_issues"] = has_issues
        if has_projects:
            data["has_projects"] = has_projects
        if has_wiki:
            data["has_wiki"] = has_wiki
        if is_template:
            data["is_template"] = is_template
        if team_id:
            data["team_id"] = team_id
        if auto_init:
            data["auto_init"] = auto_init
        if gitignore_template:
            data["gitignore_template"] = gitignore_template
        if license_template:
            data["license_template"] = license_template
        if allow_squash_merge:
            data["allow_squash_merge"] = allow_squash_merge
        if allow_merge_commit:
            data["allow_merge_commit"] = allow_merge_commit
        if allow_rebase_merge:
            data["allow_rebase_merge "] = allow_rebase_merge
        if allow_auto_merge:
            data["allow_auto_merge"] = allow_auto_merge
        if delete_branch_on_merge:
            data["delete_branch_on_merge"] = delete_branch_on_merge
        if use_squash_pr_title_as_default:
            data["use_squash_pr_title_as_default"] = use_squash_pr_title_as_default

        return await self.request("POST", f"/orgs/{org}/repos", json=data)

    async def get_repo(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}")

    async def update_repo(
        self,
        *,
        owner: str,
        repo: str,
        name: Optional[str] = None,
        description: Optional[str] = None,
        homepage: Optional[str] = None,
        private: Optional[bool] = None,
        visibility: Optional[Literal["public", "private", "internal"]] = None,
        security_and_analysis: Optional[SecurityAndAnalysis] = None,
        has_issues: Optional[bool] = None,
        has_projects: Optional[bool] = None,
        has_wiki: Optional[bool] = None,
        is_template: Optional[bool] = None,
        default_branch: Optional[str] = None,
        allow_squash_merge: Optional[bool] = None,
        allow_merge_commit: Optional[bool] = None,
        allow_rebase_merge: Optional[bool] = None,
        allow_auto_merge: Optional[bool] = None,
        delete_branch_on_merge: Optional[bool] = None,
        use_squash_pr_title_as_default: Optional[bool] = None,
        archived: Optional[bool] = None,
        allow_forking: Optional[bool] = None,
    ):
        data = {}

        if name:
            data["name"] = name
        if description:
            data["description"] = description
        if homepage:
            data["homepage"] = homepage
        if private:
            data["private"] = private
        if visibility:
            data["visibility"] = visibility
        if security_and_analysis:
            data["security_and_analysis"] = security_and_analysis
        if has_issues:
            data["has_issues"] = has_issues
        if has_projects:
            data["has_projects"] = has_projects
        if has_wiki:
            data["has_wiki"] = has_wiki
        if is_template:
            data["is_template"] = is_template
        if default_branch:
            data["default_branch"] = default_branch
        if allow_squash_merge:
            data["allow_squash_merge"] = allow_squash_merge
        if allow_merge_commit:
            data["allow_merge_commit"] = allow_merge_commit
        if allow_rebase_merge:
            data["allow_rebase_merge "] = allow_rebase_merge
        if allow_auto_merge:
            data["allow_auto_merge "] = allow_auto_merge
        if delete_branch_on_merge:
            data["delete_branch_on_merge "] = delete_branch_on_merge
        if use_squash_pr_title_as_default:
            data["use_squash_pr_title_as_default"] = use_squash_pr_title_as_default
        if archived:
            data["archived"] = archived
        if allow_forking:
            data["allow_forking"] = allow_forking

        return await self.request("PATCH", f"/repos/{owner}/{repo}", json=data)

    async def delete_repo(self, *, owner: str, repo: str):
        return await self.request("DELETE", f"/repos/{owner}/{repo}")

    async def enable_automated_security_fixes_for_repo(self, *, owner: str, repo: str):
        return await self.request("PUT", f"/repos/{owner}/{repo}/automated-security-fixes")

    async def disable_automated_security_fixes_for_repo(self, *, owner: str, repo: str):
        return await self.request("DELETE", f"/repos/{owner}/{repo}/automated-security-fixes")

    async def list_codeowners_errors_for_repo(
        self, *, owner: str, repo: str, ref: Optional[str] = None
    ):
        params = {}

        if ref:
            params["ref"] = ref

        return await self.request("GET", f"/repos/{owner}/{repo}/codeowners/errors", params=params)

    async def list_repo_contributors(
        self,
        *,
        owner: str,
        repo: str,
        anon: Optional[bool] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if anon:
            params["anon"] = anon
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/contributors", params=params)

    async def create_repo_dispatch_event(
        self, *, owner: str, repo: str, event_name: str, client_payload: Optional[str] = None
    ):
        data = {
            "event_name": event_name,
        }

        if client_payload:
            data["client_payload"] = client_payload

        return await self.request("POST", f"/repos/{owner}/{repo}/dispatches", json=data)

    async def list_repo_languages(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/languages")

    async def list_repo_tags(
        self, *, owner: str, repo: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/tags", params=params)

    async def list_repo_teams(
        self, *, owner: str, repo: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/teams", params=params)

    async def get_all_repo_topics(
        self, *, owner: str, repo: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/topics", params=params)

    async def replace_all_repo_topics(self, *, owner: str, repo: str, names: List[str]):
        return await self.request("PUT", f"/repos/{owner}/{repo}/topics", json={"names": names})

    async def transfer_repo(
        self, *, owner: str, repo: str, new_owner: str, team_ids: Optional[List[int]] = None
    ):
        data: Dict[str, Union[str, List[int]]] = {
            "new_owner": new_owner,
        }

        if team_ids:
            data["team_ids"] = team_ids

        return await self.request("POST", f"/repos/{owner}/{repo}/transfer", json=data)

    async def check_vulnerability_alerts_enabled_for_repo(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/vulnerability-alerts")

    async def enable_repo_vulnerability_alerts(self, *, owner: str, repo: str):
        return await self.request("PUT", f"/repos/{owner}/{repo}/vulnerability-alerts")

    async def disable_repo_vulnerability_alerts(self, *, owner: str, repo: str):
        return await self.request("DELETE", f"/repos/{owner}/{repo}/vulnerability-alerts")

    async def create_repo_using_template(
        self,
        *,
        template_owner: str,
        template_repo: str,
        owner: Optional[str] = None,
        name: str,
        include_all_branches: Optional[bool] = None,
        private: Optional[bool] = None,
    ):
        data: Dict[str, Union[str, bool]] = {
            "name": name,
        }

        if owner:
            data["owner"] = owner
        if include_all_branches:
            data["include_all_branches"] = include_all_branches
        if private:
            data["private"] = private

        return await self.request(
            "POST", f"/repos/{template_owner}/{template_repo}/generate", json=data
        )

    async def list_public_repos(self, *, since: Optional[int] = None):
        params = {}

        if since:
            params["since"] = since

        return await self.request("GET", "/repositories", params=params)

    async def list_repos_for_authenticated_user(
        self,
        *,
        visibility: Optional[Literal["all", "private", "public"]] = None,
        affiliation: Optional[Literal["owner", "collaborator", "organization_member"]] = None,
        type: Optional[Literal["all", "owner", "public", "private", "member"]] = None,
        sort: Optional[Literal["created", "updated", "pushed", "full_name"]] = None,
        direction: Optional[Literal["asc", "desc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
        since: Optional[str] = None,
        before: Optional[str] = None,
    ):
        data = {}

        if visibility:
            data["visibility"] = visibility
        if affiliation:
            data["affiliation"] = affiliation
        if type:
            data["type"] = type
        if sort:
            data["sort"] = sort
        if direction:
            data["direction"] = direction
        if per_page:
            data["per_page"] = per_page
        if page:
            data["page"] = page
        if since:
            data["since"] = since
        if before:
            data["before"] = before

        return self.request("POST", "/user/repos", json=data)

    async def create_repo_for_authenticated_user(
        self,
        *,
        name: str,
        description: Optional[str] = None,
        homepage: Optional[str] = None,
        private: Optional[bool] = None,
        has_issues: Optional[bool] = None,
        has_projects: Optional[bool] = None,
        has_wiki: Optional[bool] = None,
        team_id: Optional[int] = None,
        auto_init: Optional[bool] = None,
        gitignore_template: Optional[str] = None,
        license_template: Optional[str] = None,
        allow_squash_merge: Optional[bool] = None,
        allow_merge_commit: Optional[bool] = None,
        allow_rebase_merge: Optional[bool] = None,
        allow_auto_merge: Optional[bool] = None,
        delete_branch_on_merge: Optional[bool] = None,
        has_downloads: Optional[bool] = None,
        is_template: Optional[bool] = None,
    ):
        data: Dict[str, Union[str, bool, int]] = {
            "name": name,
        }

        if description:
            data["description"] = description
        if homepage:
            data["homepage"] = homepage
        if private:
            data["private"] = private
        if has_issues:
            data["has_issues"] = has_issues
        if has_projects:
            data["has_projects"] = has_projects
        if has_wiki:
            data["has_wiki"] = has_wiki
        if team_id:
            data["team_id"] = team_id
        if auto_init:
            data["auto_init"] = auto_init
        if gitignore_template:
            data["gitignore_template"] = gitignore_template
        if license_template:
            data["license_template"] = license_template
        if allow_squash_merge:
            data["allow_squash_merge"] = allow_squash_merge
        if allow_merge_commit:
            data["allow_merge_commit"] = allow_merge_commit
        if allow_rebase_merge:
            data["allow_rebase_merge"] = allow_rebase_merge
        if allow_auto_merge:
            data["allow_auto_merge"] = allow_auto_merge
        if delete_branch_on_merge:
            data["delete_branch_on_merge"] = delete_branch_on_merge
        if has_downloads:
            data["has_downloads"] = has_downloads
        if is_template:
            data["is_template"] = is_template

        return await self.request("POST", "/user/repos", json=data)

    async def list_repos_for_user(
        self,
        *,
        username: str,
        type: Optional[
            Literal["all", "public", "private", "forks", "sources", "member", "internal"]
        ] = None,
        sort: Optional[Literal["created", "updated", "pushed", "full_name"]] = None,
        direction: Optional[Literal["asc", "desc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if type:
            params["type"] = type
        if sort:
            params["sort"] = sort
        if direction:
            params["direction"] = direction
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/repos", params=params)

    async def list_repo_autolinks(self, *, owner: str, repo: str, page: Optional[int] = None):
        params = {}

        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/autolinks", params=params)

    async def create_autolink_reference_for_repo(
        self, *, owner: str, repo: str, key_prefix: str, url_template: str
    ):
        return await self.request(
            "POST",
            f"/repos/{owner}/{repo}/autolinks",
            json={"key_prefix": key_prefix, "url_template": url_template},
        )

    async def get_autolink_reference_for_repo(self, *, owner: str, repo: str, autolink_id: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/autolinks/{autolink_id}")

    async def delete_autolink_reference_for_repo(self, *, owner: str, repo: str, autolink_id: str):
        return await self.request("DELETE", f"/repos/{owner}/{repo}/autolinks/{autolink_id}")

    async def get_repo_content(
        self, *, owner: str, repo: str, path: str, ref: Optional[str] = None
    ):
        params = {}

        if ref:
            params["ref"] = ref

        return await self.request("GET", f"/repos/{owner}/{repo}/contents/{path}", params=params)

    async def create_or_update_repo_file_contents(
        self,
        *,
        owner: str,
        repo: str,
        path: str,
        message: str,
        content: str,
        sha: Optional[str] = None,
        branch: Optional[str] = None,
        committer: Optional[Committer] = None,
        author: Optional[Author] = None,
    ):
        data: Dict[str, Union[str, Committer, Author]] = {
            "message": message,
            "content": content,
        }

        if sha:
            data["sha"] = sha
        if branch:
            data["branch"] = branch
        if committer:
            data["committer"] = committer
        if author:
            data["author"] = author

        return await self.request("PUT", f"/repos/{owner}/{repo}/contents/{path}", json=data)

    async def delete_repo_file(
        self,
        *,
        owner: str,
        repo: str,
        path: str,
        message: str,
        sha: str,
        branch: Optional[str] = None,
        committer: Optional[OptionalCommitter] = None,
        author: Optional[OptionalAuthor] = None,
    ):
        data: Dict[str, Union[str, OptionalCommitter, OptionalAuthor]] = {
            "message": message,
            "sha": sha,
        }

        if branch:
            data["branch"] = branch
        if committer:
            data["committer"] = committer
        if author:
            data["author"] = author

        return await self.request("DELETE", f"/repos/{owner}/{repo}/contents/{path}", json=data)

    async def get_repo_readme(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/readme")

    async def get_repo_readme_for_directory(
        self, *, owner: str, repo: str, dir: str, ref: Optional[str] = None
    ):
        params = {}

        if ref:
            params["ref"] = ref

        return await self.request("GET", f"/repos/{owner}/{repo}/readme/{dir}", params=params)

    async def download_repo_archive(
        self,
        *,
        owner: str,
        repo: str,
        archive_format: Literal["tarball", "zipball"],
        ref: Optional[str] = None,
    ):
        params = {}

        if ref:
            params["ref"] = ref

        return await self.request("GET", f"/repos/{owner}/{repo}/{archive_format}", params=params)

    async def list_repo_forks(
        self,
        *,
        owner: str,
        repo: str,
        sort: Optional[Literal["newest", "oldest", "stargazers", "watchers"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if sort:
            params["sort"] = sort
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/forks", params=params)

    async def create_repo_fork(self, *, owner: str, repo: str, organization: Optional[str] = None):
        data = {}

        if organization:
            data["organization"] = organization

        return await self.request("POST", f"/repos/{owner}/{repo}/forks", json=data)

    async def enable_git_lfs_for_repo(self, *, owner: str, repo: str):
        return await self.request("PUT", f"/repos/{owner}/{repo}/lfs")

    async def disable_git_lfs_for_repo(self, *, owner: str, repo: str):
        return await self.request("DELETE", f"/repos/{owner}/{repo}/lfs")

    async def list_tag_protection_states_for_repo(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/tags/protection")

    async def create_tag_protection_state_for_repo(
        self,
        *,
        owner: str,
        repo: str,
        pattern: str,
    ):
        data = {}

        if pattern:
            data["pattern"] = pattern

        return await self.request("POST", f"/repos/{owner}/{repo}/tags/protection", json=data)

    async def delete_tag_protection_state_for_repo(
        self, *, owner: str, repo: str, tag_protection_id: int
    ):
        return await self.request(
            "DELETE", f"/repos/{owner}/{repo}/tags/protection/{tag_protection_id}"
        )

    # === GISTS === #

    async def list_gists_for_authenticated_user(
        self,
        *,
        since: Optional[str] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if since:
            params["since"] = since
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/gists", params=params)

    async def create_gist(
        self, *, description: Optional[str] = None, files: List[File], public: Optional[bool] = None
    ):
        data: Dict[str, Union[str, bool, Dict[str, str]]] = {
            "files": {f.name: f.read() for f in files},
        }

        if description:
            data["description"] = description
        if public:
            data["public"] = public

        return await self.request("POST", "/gists", json=data)

    async def list_public_gists(
        self,
        *,
        since: Optional[str] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if since:
            params["since"] = since
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/gists/public", params=params)

    async def list_starred_gists(
        self,
        *,
        since: Optional[str] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if since:
            params["since"] = since
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/gists/starred", params=params)

    async def get_gist(self, *, gist_id: str):
        return await self.request("GET", f"/gists/{gist_id}")

    async def update_gist(
        self, *, gist_id: str, description: Optional[str] = None, files: Optional[List[File]] = None
    ):
        data = {}

        if description:
            data["description"] = description
        if files:
            data["files"] = {f.name: f.read() for f in files}

        return await self.request("PATCH", f"/gists/{gist_id}")

    async def delete_gist(self, *, gist_id: str):
        return await self.request("DELETE", f"/gists/{gist_id}")

    async def list_gist_commits(
        self, *, gist_id: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/gists/{gist_id}/commits", params=params)

    async def list_gist_forks(
        self, *, gist_id: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/gists/{gist_id}/forks", params=params)

    async def fork_gist(self, *, gist_id: str):
        return await self.request("POST", f"/gists/{gist_id}/forks")

    async def check_gist_starred(self, *, gist_id: str):
        return await self.request("GET", f"/gists/{gist_id}/star")

    async def star_gist(self, *, gist_id: str):
        return await self.request("PUT", f"/gists/{gist_id}/star")

    async def unstar_gist(self, *, gist_id: str):
        return await self.request("DELETE", f"/gists/{gist_id}/star")

    async def get_gist_revision(self, *, gist_id: str, sha: str):
        return await self.request("GET", f"/gists/{gist_id}/{sha}")

    async def list_gists_for_user(
        self,
        *,
        username: str,
        since: Optional[str] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {}

        if since:
            params["since"] = since
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/users/{username}/gists", params=params)

    async def list_gist_comments(
        self, *, gist_id: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/gists/{gist_id}/comments", params=params)

    async def create_gist_comment(self, *, gist_id: str, body: str):
        return await self.request("POST", f"/gists/{gist_id}/comments", json={"body": body})

    async def get_gist_comment(self, *, gist_id: str, comment_id: str):
        return await self.request("GET", f"/gists/{gist_id}/comments/{comment_id}")

    async def update_gist_comment(self, *, gist_id: str, comment_id: str, body: str):
        return await self.request(
            "PATCH", f"/gists/{gist_id}/comments/{comment_id}", json={"body": body}
        )

    async def delete_gist_comment(self, *, gist_id: str, comment_id: str):
        return await self.request("DELETE", f"/gists/{gist_id}/comments/{comment_id}")

    # === LICENSES === #

    async def get_all_commonly_used_licenses(self):
        return await self.request("GET", "/licenses")

    async def get_license(self, *, license: str):
        return await self.request("GET", f"/licenses/{license}")

    async def get_license_for_repo(self, *, owner: str, repo: str):
        return await self.request("GET", f"/repos/{owner}/{repo}/license")

    # === GITIGNORE === #

    async def get_all_gitignore_templates(self):
        return await self.request("GET", "/gitignore/templates")

    async def get_gitignore_template(self, *, name: str):
        return await self.request("GET", f"/gitignore/templates/{name}")

    # === EMOJIS === #

    async def get_emojis(self):
        return await self.request("GET", "/emojis")

    # === CODES OF CONDUCT === #

    async def get_all_codes_of_conduct(self):
        return await self.request("GET", "/codes_of_conduct")

    async def get_code_of_conduct(self, *, key: str):
        return await self.request("GET", f"/codes_of_conduct/{key}")

    # === DEPLOY KEYS === #

    async def list_deploy_keys(
        self, *, owner: str, repo: str, per_page: Optional[int] = None, page: Optional[int] = None
    ):
        params = {}

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", f"/repos/{owner}/{repo}/keys", params=params)

    async def create_deploy_key(
        self,
        *,
        owner: str,
        repo: str,
        title: Optional[str] = None,
        key: str,
        read_only: Optional[bool] = None,
    ):
        data: Dict[str, Union[str, bool]] = {
            "key": key,
        }

        if title:
            data["title"] = title
        if read_only:
            data["read_only"] = read_only

        return await self.request("POST", f"/repos/{owner}/{repo}/keys", data=data)

    async def get_deploy_key(self, *, owner: str, repo: str, key_id: int):
        return await self.request("GET", f"/repos/{owner}/{repo}/keys/{key_id}")

    async def delete_deploy_key(self, *, owner: str, repo: str, key_id: int):
        return await self.request("DELETE", f"/repos/{owner}/{repo}/keys/{key_id}")

    # === MARKDOWN === #

    async def render_markdown(
        self,
        *,
        text: str,
        mode: Optional[Literal["markdown", "gfm"]] = None,
        context: Optional[str] = None,
    ):
        data = {
            "text": text,
        }

        if mode:
            data["mode"] = mode
        if context:
            data["context"] = context

        return await self.request("POST", "/markdown", data=data)

    # TODO: Implement Markdown raw request, idk

    # === META === #

    async def get_github_api_root(self):
        return await self.request("GET", "/")

    async def get_git_hub_meta_info(self):
        return await self.request("GET", "/meta")

    async def get_octocat(self):
        return await self.request("GET", "/octocat")

    async def get_the_zen_of_github(self):
        return await self.request("GET", "/zen")

    # === SEARCH === #

    async def search_code(
        self,
        *,
        q: str,
        sort: Optional[Literal["indexed"]] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        data: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if sort:
            data["sort"] = sort
        if order:
            data["order"] = order
        if per_page:
            data["per_page"] = per_page
        if page:
            data["page"] = page

        return await self.request("GET", "/search/code", data=data)

    async def search_commits(
        self,
        *,
        q: str,
        sort: Optional[Literal["author-date", "committer-date"]] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/commits", params=params)

    async def search_issues_and_pull_requests(
        self,
        *,
        q: str,
        sort: Optional[
            Literal[
                "comments",
                "reactions",
                "reactions-+1",
                "reactions--1",
                "reactions-smile",
                "reactions-thinking_face",
                "reactions-heart",
                "reactions-tada",
                "interactions",
                "created",
                "updated",
            ]
        ] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/issues", params=params)

    async def search_labels(
        self,
        *,
        repository_id: int,
        q: str,
        sort: Optional[Literal["created", "updated"]] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params = {
            "repository_id": repository_id,
            "q": q,
        }

        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/labels", params=params)

    async def search_repos(
        self,
        *,
        q: str,
        sort: Optional[Literal["stars", "forks", "help-wanted-issues", "updated"]] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/repositories", params=params)

    async def search_topics(
        self,
        *,
        q: str,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/topics", params=params)

    async def search_users(
        self,
        *,
        q: str,
        sort: Optional[Literal["followers", "repositories", "joined"]] = None,
        order: Optional[Literal["desc", "asc"]] = None,
        per_page: Optional[int] = None,
        page: Optional[int] = None,
    ):
        params: Dict[str, Union[str, int]] = {
            "q": q,
        }

        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        if per_page:
            params["per_page"] = per_page
        if page:
            params["page"] = page

        return await self.request("GET", "/search/users", params=params)