<commit_before>"""Highlevel API for managing PRs on Github"""

import abc
import datetime
import logging
import time

from copy import copy
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple

import aiohttp
import backoff
import cachetools
import gidgethub
import gidgethub.aiohttp
import gidgethub.sansio
import jwt


logger = logging.getLogger(__name__)  # pylint: disable=invalid-name


#: State for Github Issues
IssueState = Enum("IssueState", "open closed all")  # pylint: disable=invalid-name

#: State of Github Check Run
CheckRunStatus = Enum("CheckRunState", "queued in_progress completed")

#: Conclusion of Github Check Run
CheckRunConclusion = Enum("CheckRunConclusion",
                          "success failure neutral cancelled timed_out action_required")


def iso_now() -> str:
    """Creates ISO 8601 timestamp in format
    ``YYYY-MM-DDTHH:MM:SSZ`` as required by Github
    """
    return datetime.datetime.utcnow().replace(microsecond=0).isoformat()+'Z'


class GitHubHandler:
    """Handles interaction with GitHub

    Arguments:
      token: OAUTH token granting permissions to GH
      dry_run: Don't actually modify things if set
      to_user: Target User/Org for PRs
      to_repo: Target repository within **to_user**
    """
    PULLS = "/repos/{user}/{repo}/pulls{/number}{?head,base,state}"
    PULL_FILES = "/repos/{user}/{repo}/pulls/{number}/files"
    PULL_COMMITS = "/repos/{user}/{repo}/pulls/{number}/commits"
    ISSUES = "/repos/{user}/{repo}/issues{/number}"
    ISSUE_COMMENTS = "/repos/{user}/{repo}/issues/{number}/comments"
    COMMENTS = "/repos/{user}/{repo}/issues/comments{/comment_id}"
    ORG_MEMBERS = "/orgs/{user}/members{/username}"
    CHECK_RUN = "/repos/{user}/{repo}/check-runs{/id}"
    GET_CHECK_RUNS = "/repos/{user}/{repo}/commits/{commit}/check-runs"

    STATE = IssueState

    def __init__(self, token: str = None,
                 dry_run: bool = False,
                 to_user: str = "bioconda",
                 to_repo: str = "bioconda-recipes",
                 installation: int = None) -> None:
        self.token = token
        self.dry_run = dry_run
        self.installation = installation
        self.user = to_user
        self.repo = to_repo
        self.var_default = {'user': to_user,
                            'repo': to_repo}

        # filled in by login():
        self.api: gidgethub.abc.GitHubAPI = None
        self.username: str = None

    def __str__(self):
        return f"{self.__class__.__name__}({self.user}/{self.repo})"

    def __repr__(self):
        return str(self)

    def for_json(self):
        return {
            '__module__': self.__module__,
            '__type__': self.__class__.__qualname__,
            'dry_run': self.dry_run,
            'to_user': self.user,
            'to_repo': self.repo,
            'installation': self.installation
        }

    def __to_json__(self):
        return self.for_json()

    @property
    def rate_limit(self) -> gidgethub.sansio.RateLimit:
        """Last recorded rate limit data"""
        return self.api.rate_limit

    def set_oauth_token(self, token: str) -> None:
        """Update oauth token for the wrapped GitHubAPI object"""
        self.api.oauth_token = token

    @abc.abstractmethod
    def create_api_object(self, *args, **kwargs):
        """Create API object"""

    def get_file_relurl(self, path: str, branch_name: str = "master") -> str:
        """Format domain relative url for **path** on **branch_name**"""
        return "/{user}/{repo}/tree/{branch_name}/{path}".format(
            branch_name=branch_name, path=path, **self.var_default)

    async def login(self, *args, **kwargs):
        """Log into API (fills `self.username`)"""

        self.create_api_object(*args, **kwargs)

        if not self.token:
            self.username = "UNKNOWN [no token]"
        else:
            user = await self.api.getitem("/user")
            self.username = user["login"]

    async def is_member(self, username) -> bool:
        """Check if **username** is member of current org"""
        if not username:
            return False
        var_data = copy(self.var_default)
        var_data['username'] = username
        try:
            await self.api.getitem(self.ORG_MEMBERS, var_data)
        except gidgethub.BadRequest:
            logger.debug("User %s is not a member of %s", username, var_data['user'])
            return False
        logger.debug("User %s IS a member of %s", username, var_data['user'])
        return True

    # pylint: disable=too-many-arguments
    @backoff.on_exception(backoff.fibo, gidgethub.BadRequest, max_tries=14)
    async def get_prs(self,
                      from_branch: Optional[str] = None,
                      from_user: Optional[str] = None,
                      to_branch: Optional[str] = None,
                      number: Optional[int] = None,
                      state: Optional[IssueState] = None) -> List[Dict[Any, Any]]:
        """Retrieve list of PRs matching parameters

        Arguments:
          from_branch: Name of branch from which PR asks to pull
          from_user: Name of user/org in from which to pull
                     (default: from auth)
          to_branch: Name of branch into which to pull (default: master)
          number: PR number
        """
        var_data = copy(self.var_default)
        if not from_user:
            from_user = self.username
        if from_branch:
            if from_user:
                var_data['head'] = f"{from_user}:{from_branch}"
            else:
                var_data['head'] = from_branch
        if to_branch:
            var_data['base'] = to_branch
        if number:<commit_msg>Githandler: lower get_prs retry time<commit_after>"""Highlevel API for managing PRs on Github"""

import abc
import datetime
import logging
import time

from copy import copy
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple

import aiohttp
import backoff
import cachetools
import gidgethub
import gidgethub.aiohttp
import gidgethub.sansio
import jwt


logger = logging.getLogger(__name__)  # pylint: disable=invalid-name


#: State for Github Issues
IssueState = Enum("IssueState", "open closed all")  # pylint: disable=invalid-name

#: State of Github Check Run
CheckRunStatus = Enum("CheckRunState", "queued in_progress completed")

#: Conclusion of Github Check Run
CheckRunConclusion = Enum("CheckRunConclusion",
                          "success failure neutral cancelled timed_out action_required")


def iso_now() -> str:
    """Creates ISO 8601 timestamp in format
    ``YYYY-MM-DDTHH:MM:SSZ`` as required by Github
    """
    return datetime.datetime.utcnow().replace(microsecond=0).isoformat()+'Z'


class GitHubHandler:
    """Handles interaction with GitHub

    Arguments:
      token: OAUTH token granting permissions to GH
      dry_run: Don't actually modify things if set
      to_user: Target User/Org for PRs
      to_repo: Target repository within **to_user**
    """
    PULLS = "/repos/{user}/{repo}/pulls{/number}{?head,base,state}"
    PULL_FILES = "/repos/{user}/{repo}/pulls/{number}/files"
    PULL_COMMITS = "/repos/{user}/{repo}/pulls/{number}/commits"
    ISSUES = "/repos/{user}/{repo}/issues{/number}"
    ISSUE_COMMENTS = "/repos/{user}/{repo}/issues/{number}/comments"
    COMMENTS = "/repos/{user}/{repo}/issues/comments{/comment_id}"
    ORG_MEMBERS = "/orgs/{user}/members{/username}"
    CHECK_RUN = "/repos/{user}/{repo}/check-runs{/id}"
    GET_CHECK_RUNS = "/repos/{user}/{repo}/commits/{commit}/check-runs"

    STATE = IssueState

    def __init__(self, token: str = None,
                 dry_run: bool = False,
                 to_user: str = "bioconda",
                 to_repo: str = "bioconda-recipes",
                 installation: int = None) -> None:
        self.token = token
        self.dry_run = dry_run
        self.installation = installation
        self.user = to_user
        self.repo = to_repo
        self.var_default = {'user': to_user,
                            'repo': to_repo}

        # filled in by login():
        self.api: gidgethub.abc.GitHubAPI = None
        self.username: str = None

    def __str__(self):
        return f"{self.__class__.__name__}({self.user}/{self.repo})"

    def __repr__(self):
        return str(self)

    def for_json(self):
        return {
            '__module__': self.__module__,
            '__type__': self.__class__.__qualname__,
            'dry_run': self.dry_run,
            'to_user': self.user,
            'to_repo': self.repo,
            'installation': self.installation
        }

    def __to_json__(self):
        return self.for_json()

    @property
    def rate_limit(self) -> gidgethub.sansio.RateLimit:
        """Last recorded rate limit data"""
        return self.api.rate_limit

    def set_oauth_token(self, token: str) -> None:
        """Update oauth token for the wrapped GitHubAPI object"""
        self.api.oauth_token = token

    @abc.abstractmethod
    def create_api_object(self, *args, **kwargs):
        """Create API object"""

    def get_file_relurl(self, path: str, branch_name: str = "master") -> str:
        """Format domain relative url for **path** on **branch_name**"""
        return "/{user}/{repo}/tree/{branch_name}/{path}".format(
            branch_name=branch_name, path=path, **self.var_default)

    async def login(self, *args, **kwargs):
        """Log into API (fills `self.username`)"""

        self.create_api_object(*args, **kwargs)

        if not self.token:
            self.username = "UNKNOWN [no token]"
        else:
            user = await self.api.getitem("/user")
            self.username = user["login"]

    async def is_member(self, username) -> bool:
        """Check if **username** is member of current org"""
        if not username:
            return False
        var_data = copy(self.var_default)
        var_data['username'] = username
        try:
            await self.api.getitem(self.ORG_MEMBERS, var_data)
        except gidgethub.BadRequest:
            logger.debug("User %s is not a member of %s", username, var_data['user'])
            return False
        logger.debug("User %s IS a member of %s", username, var_data['user'])
        return True

    # pylint: disable=too-many-arguments
    @backoff.on_exception(backoff.fibo, gidgethub.BadRequest, max_tries=10)
    async def get_prs(self,
                      from_branch: Optional[str] = None,
                      from_user: Optional[str] = None,
                      to_branch: Optional[str] = None,
                      number: Optional[int] = None,
                      state: Optional[IssueState] = None) -> List[Dict[Any, Any]]:
        """Retrieve list of PRs matching parameters

        Arguments:
          from_branch: Name of branch from which PR asks to pull
          from_user: Name of user/org in from which to pull
                     (default: from auth)
          to_branch: Name of branch into which to pull (default: master)
          number: PR number
        """
        var_data = copy(self.var_default)
        if not from_user:
            from_user = self.username
        if from_branch:
            if from_user:
                var_data['head'] = f"{from_user}:{from_branch}"
            else:
                var_data['head'] = from_branch
        if to_branch:
            var_data['base'] = to_branch
        if number:
{'fraction_non_alphanumeric': 0.11786741005314051, 'fraction_numerical': 0.0012196184336614687, 'mean_word_length': 2.389430174195453, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '221520', 'n_tokens_mistral': 3492, 'n_tokens_neox': 3371, 'n_words': 883}