From b6258f4be9bee07beb0935a749289b152b20f7f0 Mon Sep 17 00:00:00 2001 From: Trim21 Date: Fri, 16 Jun 2023 23:57:35 +0800 Subject: [PATCH] merge GithubObject.pyi/Requester.pyi stubs back to source (#2463) Co-authored-by: Enrico Minack Co-authored-by: Jonathan Leitschuh --- github/AccessToken.py | 43 ++--- github/AccessToken.pyi | 29 ---- github/Authorization.py | 127 +++++++------- github/Authorization.pyi | 37 ---- github/GithubException.py | 6 +- github/GithubObject.py | 192 ++++++++++++++------- github/GithubObject.pyi | 134 --------------- github/RepositoryAdvisory.py | 18 +- github/Requester.py | 323 +++++++++++++++++++++++++---------- github/Requester.pyi | 232 ------------------------- 10 files changed, 461 insertions(+), 680 deletions(-) delete mode 100644 github/AccessToken.pyi delete mode 100644 github/Authorization.pyi delete mode 100644 github/GithubObject.pyi delete mode 100644 github/Requester.pyi diff --git a/github/AccessToken.py b/github/AccessToken.py index 98169fa2de..7cea4e866f 100644 --- a/github/AccessToken.py +++ b/github/AccessToken.py @@ -21,16 +21,27 @@ ################################################################################ from datetime import datetime, timedelta, timezone +from typing import Optional import github.GithubObject +from .GithubObject import Attribute + class AccessToken(github.GithubObject.NonCompletableGithubObject): """ This class represents access tokens. """ - def __repr__(self): + _created: datetime + _token: Attribute[str] + _type: Attribute[str] + _scope: Attribute[str] + _expires_in: Attribute[Optional[int]] + _refresh_token: Attribute[str] + _refresh_expires_in: Attribute[Optional[int]] + + def __repr__(self) -> str: return self.get__repr__( { "token": f"{self.token[:5]}...", @@ -45,44 +56,42 @@ def __repr__(self): ) @property - def token(self): + def token(self) -> str: """ :type: string """ return self._token.value @property - def type(self): + def type(self) -> str: """ :type: string """ return self._type.value @property - def scope(self): + def scope(self) -> str: """ :type: string """ return self._scope.value @property - def created(self): + def created(self) -> datetime: """ :type: datetime """ return self._created @property - def expires_in(self): + def expires_in(self) -> Optional[int]: """ :type: Optional[int] """ - if self._expires_in is not github.GithubObject.NotSet: - return self._expires_in.value - return None + return self._expires_in.value @property - def expires_at(self): + def expires_at(self) -> Optional[datetime]: """ :type: Optional[datetime] """ @@ -92,25 +101,21 @@ def expires_at(self): return None @property - def refresh_token(self): + def refresh_token(self) -> Optional[str]: """ :type: Optional[string] """ - if self._refresh_token is not github.GithubObject.NotSet: - return self._refresh_token.value - return None + return self._refresh_token.value @property - def refresh_expires_in(self): + def refresh_expires_in(self) -> Optional[int]: """ :type: Optional[int] """ - if self._refresh_expires_in is not github.GithubObject.NotSet: - return self._refresh_expires_in.value - return None + return self._refresh_expires_in.value @property - def refresh_expires_at(self): + def refresh_expires_at(self) -> Optional[datetime]: """ :type: Optional[datetime] """ diff --git a/github/AccessToken.pyi b/github/AccessToken.pyi deleted file mode 100644 index a0f959f942..0000000000 --- a/github/AccessToken.pyi +++ /dev/null @@ -1,29 +0,0 @@ -import datetime -from typing import Any, Dict, Optional - -from github.Auth import AppUserAuth -from github.GithubObject import NonCompletableGithubObject - -class AccessToken(NonCompletableGithubObject): - def __repr__(self) -> str: ... - def _initAttributes(self) -> None: ... - def _useAttributes(self, attributes: Dict[str, Any]) -> None: ... - @property - def token(self) -> str: ... - @property - def type(self) -> str: ... - @property - def scope(self) -> str: ... - @property - def created(self) -> datetime.datetime: ... - @property - def expires_in(self) -> Optional[int]: ... - @property - def expires_at(self) -> Optional[datetime.datetime]: ... - @property - def refresh_token(self) -> Optional[str]: ... - @property - def refresh_expires_in(self) -> Optional[int]: ... - @property - def refresh_expires_at(self) -> Optional[datetime.datetime]: ... - def as_app_user_auth(self) -> AppUserAuth: ... diff --git a/github/Authorization.py b/github/Authorization.py index b495d4b324..5fe99f865f 100644 --- a/github/Authorization.py +++ b/github/Authorization.py @@ -27,9 +27,15 @@ # along with PyGithub. If not, see . # # # ################################################################################ +from datetime import datetime +from typing import TYPE_CHECKING, List, Optional import github.AuthorizationApplication import github.GithubObject +from github.GithubObject import Attribute, NotSet, Opt, _NotSetType + +if TYPE_CHECKING: + from github.AuthorizationApplication import AuthorizationApplication class Authorization(github.GithubObject.CompletableGithubObject): @@ -37,19 +43,26 @@ class Authorization(github.GithubObject.CompletableGithubObject): This class represents Authorizations. The reference can be found here https://docs.github.com/en/enterprise-server@3.0/rest/reference/oauth-authorizations """ - def __repr__(self): + _app: Attribute["AuthorizationApplication"] + _created_at: Attribute[datetime] + _id: Attribute[int] + _note: Attribute[Optional[str]] + _note_url: Attribute[Optional[str]] + _scopes: Attribute[str] + _token: Attribute[str] + _updated_at: Attribute[datetime] + _url: Attribute[str] + + def __repr__(self) -> str: return self.get__repr__({"scopes": self._scopes.value}) @property - def app(self): - """ - :type: :class:`github.AuthorizationApplication.AuthorizationApplication` - """ + def app(self) -> "AuthorizationApplication": self._completeIfNotSet(self._app) return self._app.value @property - def created_at(self): + def created_at(self) -> datetime: """ :type: datetime.datetime """ @@ -57,76 +70,54 @@ def created_at(self): return self._created_at.value @property - def id(self): - """ - :type: integer - """ + def id(self) -> int: self._completeIfNotSet(self._id) return self._id.value @property - def note(self): - """ - :type: string - """ + def note(self) -> Optional[str]: self._completeIfNotSet(self._note) return self._note.value @property - def note_url(self): - """ - :type: string - """ + def note_url(self) -> Optional[str]: self._completeIfNotSet(self._note_url) return self._note_url.value @property - def scopes(self): - """ - :type: list of string - """ + def scopes(self) -> str: self._completeIfNotSet(self._scopes) return self._scopes.value @property - def token(self): - """ - :type: string - """ + def token(self) -> str: self._completeIfNotSet(self._token) return self._token.value @property - def updated_at(self): - """ - :type: datetime.datetime - """ + def updated_at(self) -> datetime: self._completeIfNotSet(self._updated_at) return self._updated_at.value @property - def url(self): - """ - :type: string - """ + def url(self) -> str: self._completeIfNotSet(self._url) return self._url.value - def delete(self): + def delete(self) -> None: """ :calls: `DELETE /authorizations/{id} `_ - :rtype: None """ headers, data = self._requester.requestJsonAndCheck("DELETE", self.url) def edit( self, - scopes=github.GithubObject.NotSet, - add_scopes=github.GithubObject.NotSet, - remove_scopes=github.GithubObject.NotSet, - note=github.GithubObject.NotSet, - note_url=github.GithubObject.NotSet, - ): + scopes: Opt[List[str]] = NotSet, + add_scopes: Opt[List[str]] = NotSet, + remove_scopes: Opt[List[str]] = NotSet, + note: Opt[str] = NotSet, + note_url: Opt[str] = NotSet, + ) -> None: """ :calls: `PATCH /authorizations/{id} `_ :param scopes: list of string @@ -136,45 +127,43 @@ def edit( :param note_url: string :rtype: None """ - assert scopes is github.GithubObject.NotSet or all( + assert isinstance(scopes, _NotSetType) or all( isinstance(element, str) for element in scopes ), scopes - assert add_scopes is github.GithubObject.NotSet or all( + assert isinstance(add_scopes, _NotSetType) or all( isinstance(element, str) for element in add_scopes ), add_scopes - assert remove_scopes is github.GithubObject.NotSet or all( + assert isinstance(remove_scopes, _NotSetType) or all( isinstance(element, str) for element in remove_scopes ), remove_scopes - assert note is github.GithubObject.NotSet or isinstance(note, str), note - assert note_url is github.GithubObject.NotSet or isinstance( - note_url, str - ), note_url - post_parameters = dict() - if scopes is not github.GithubObject.NotSet: - post_parameters["scopes"] = scopes - if add_scopes is not github.GithubObject.NotSet: - post_parameters["add_scopes"] = add_scopes - if remove_scopes is not github.GithubObject.NotSet: - post_parameters["remove_scopes"] = remove_scopes - if note is not github.GithubObject.NotSet: - post_parameters["note"] = note - if note_url is not github.GithubObject.NotSet: - post_parameters["note_url"] = note_url + assert isinstance(note, (_NotSetType, str)), note + assert isinstance(note_url, (_NotSetType, str)), note_url + + post_parameters = NotSet.remove_unset_items( + { + "scopes": scopes, + "add_scopes": add_scopes, + "remove_scopes": remove_scopes, + "note": note, + "note_url": note_url, + } + ) + headers, data = self._requester.requestJsonAndCheck( "PATCH", self.url, input=post_parameters ) self._useAttributes(data) def _initAttributes(self): - self._app = github.GithubObject.NotSet - self._created_at = github.GithubObject.NotSet - self._id = github.GithubObject.NotSet - self._note = github.GithubObject.NotSet - self._note_url = github.GithubObject.NotSet - self._scopes = github.GithubObject.NotSet - self._token = github.GithubObject.NotSet - self._updated_at = github.GithubObject.NotSet - self._url = github.GithubObject.NotSet + self._app = NotSet + self._created_at = NotSet + self._id = NotSet + self._note = NotSet + self._note_url = NotSet + self._scopes = NotSet + self._token = NotSet + self._updated_at = NotSet + self._url = NotSet def _useAttributes(self, attributes): if "app" in attributes: # pragma no branch diff --git a/github/Authorization.pyi b/github/Authorization.pyi deleted file mode 100644 index 059485ce87..0000000000 --- a/github/Authorization.pyi +++ /dev/null @@ -1,37 +0,0 @@ -from datetime import datetime -from typing import Any, Dict, List, Optional, Union - -from github.AuthorizationApplication import AuthorizationApplication -from github.GithubObject import CompletableGithubObject, _NotSetType - -class Authorization(CompletableGithubObject): - def __repr__(self) -> str: ... - def _initAttributes(self) -> None: ... - def _useAttributes(self, attributes: Dict[str, Any]) -> None: ... - @property - def app(self) -> AuthorizationApplication: ... - @property - def created_at(self) -> datetime: ... - def delete(self) -> None: ... - def edit( - self, - scopes: Union[_NotSetType, List[str]] = ..., - add_scopes: Union[_NotSetType, List[str]] = ..., - remove_scopes: Union[_NotSetType, List[str]] = ..., - note: Union[_NotSetType, str] = ..., - note_url: Union[_NotSetType, str] = ..., - ) -> None: ... - @property - def id(self) -> int: ... - @property - def note(self) -> Optional[str]: ... - @property - def note_url(self) -> Optional[str]: ... - @property - def scopes(self) -> List[str]: ... - @property - def token(self) -> str: ... - @property - def updated_at(self) -> datetime: ... - @property - def url(self) -> str: ... diff --git a/github/GithubException.py b/github/GithubException.py index 46fb041414..717102ca7c 100644 --- a/github/GithubException.py +++ b/github/GithubException.py @@ -42,7 +42,7 @@ class GithubException(Exception): def __init__( self, status: int, - data: Dict[str, Union[str, List[str], List[Dict[str, str]]]], + data: Union[str, Dict[str, Union[str, List[str], List[Dict[str, str]]]]], headers: Optional[Dict[str, str]], ): super().__init__() @@ -59,7 +59,9 @@ def status(self) -> int: return self.__status @property - def data(self) -> Dict[str, Union[str, List[str], List[Dict[str, str]]]]: + def data( + self, + ) -> Union[str, Dict[str, Union[str, List[str], List[Dict[str, str]]]]]: """ The (decoded) data returned by the Github API """ diff --git a/github/GithubObject.py b/github/GithubObject.py index 3ad6f8675d..6cea1cc05a 100644 --- a/github/GithubObject.py +++ b/github/GithubObject.py @@ -41,40 +41,75 @@ import datetime import typing from operator import itemgetter +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Generic, + List, + Optional, + Type, + Union, +) + +from . import Consts +from .GithubException import BadAttributeException, IncompletableObject + +if TYPE_CHECKING: + from .Requester import Requester + +T = typing.TypeVar("T") -from . import Consts, GithubException +class Attribute(Generic[T]): + @property + def value(self) -> T: + raise NotImplementedError -class _NotSetType: + +class _NotSetType(Attribute): def __repr__(self): return "NotSet" - value = None + @property + def value(self): + return None + + @staticmethod + def remove_unset_items(data: Dict[str, Any]) -> Dict[str, Any]: + return { + key: value + for key, value in data.items() + if not isinstance(value, _NotSetType) + } NotSet = _NotSetType() -T = typing.TypeVar("T") +Opt = Union[T, _NotSetType] -OptionallySet = typing.Union[T, _NotSetType] +class _ValuedAttribute(Attribute, Generic[T]): + def __init__(self, value: T): + self._value = value -class _ValuedAttribute: - def __init__(self, value): - self.value = value + @property + def value(self) -> T: + return self._value -class _BadAttribute: - def __init__(self, value, expectedType, exception=None): +class _BadAttribute(Attribute): + def __init__( + self, value: Any, expectedType: Any, exception: Optional[Exception] = None + ): self.__value = value self.__expectedType = expectedType self.__exception = exception @property def value(self): - raise GithubException.BadAttributeException( - self.__value, self.__expectedType, self.__exception - ) + raise BadAttributeException(self.__value, self.__expectedType, self.__exception) class GithubObject: @@ -86,12 +121,19 @@ class GithubObject: A global debug flag to enable header validation by requester for all objects """ CHECK_AFTER_INIT_FLAG = False + _url: Attribute[str] @classmethod - def setCheckAfterInitFlag(cls, flag): + def setCheckAfterInitFlag(cls, flag: bool) -> None: cls.CHECK_AFTER_INIT_FLAG = flag - def __init__(self, requester, headers, attributes, completed): + def __init__( + self, + requester: "Requester", + headers: Dict[str, Union[str, int]], + attributes: Any, + completed: bool, + ): self._requester = requester self._initAttributes() self._storeAndUseAttributes(headers, attributes) @@ -101,7 +143,9 @@ def __init__(self, requester, headers, attributes, completed): if self.CHECK_AFTER_INIT_FLAG: # pragma no branch (Flag always set in tests) requester.check_me(self) - def _storeAndUseAttributes(self, headers, attributes): + def _storeAndUseAttributes( + self, headers: Dict[str, Union[str, int]], attributes: Any + ) -> None: # Make sure headers are assigned before calling _useAttributes # (Some derived classes will use headers in _useAttributes) self._headers = headers @@ -109,7 +153,7 @@ def _storeAndUseAttributes(self, headers, attributes): self._useAttributes(attributes) @property - def raw_data(self): + def raw_data(self) -> Dict[str, Any]: """ :type: dict """ @@ -117,7 +161,7 @@ def raw_data(self): return self._rawData @property - def raw_headers(self): + def raw_headers(self) -> Dict[str, Union[str, int]]: """ :type: dict """ @@ -125,43 +169,45 @@ def raw_headers(self): return self._headers @staticmethod - def _parentUrl(url): + def _parentUrl(url: str) -> str: return "/".join(url.split("/")[:-1]) @staticmethod - def __makeSimpleAttribute(value, type): + def __makeSimpleAttribute(value: Any, type: Type[T]) -> Attribute[T]: if value is None or isinstance(value, type): - return _ValuedAttribute(value) + return _ValuedAttribute(value) # type: ignore else: - return _BadAttribute(value, type) + return _BadAttribute(value, type) # type: ignore @staticmethod - def __makeSimpleListAttribute(value, type): + def __makeSimpleListAttribute(value: list, type: Type[T]) -> Attribute[T]: if isinstance(value, list) and all( isinstance(element, type) for element in value ): - return _ValuedAttribute(value) + return _ValuedAttribute(value) # type: ignore else: - return _BadAttribute(value, [type]) + return _BadAttribute(value, [type]) # type: ignore @staticmethod - def __makeTransformedAttribute(value, type, transform): + def __makeTransformedAttribute( + value: T, type: Type[T], transform: Callable[[T], Any] + ) -> Attribute[T]: if value is None: - return _ValuedAttribute(None) + return _ValuedAttribute(None) # type: ignore elif isinstance(value, type): try: return _ValuedAttribute(transform(value)) except Exception as e: - return _BadAttribute(value, type, e) + return _BadAttribute(value, type, e) # type: ignore else: - return _BadAttribute(value, type) + return _BadAttribute(value, type) # type: ignore @staticmethod - def _makeStringAttribute(value): + def _makeStringAttribute(value: Optional[Union[int, str]]) -> Attribute[str]: return GithubObject.__makeSimpleAttribute(value, str) @staticmethod - def _makeIntAttribute(value): + def _makeIntAttribute(value: Optional[Union[int, str]]) -> Attribute[int]: return GithubObject.__makeSimpleAttribute(value, int) @staticmethod @@ -169,21 +215,21 @@ def _makeFloatAttribute(value): return GithubObject.__makeSimpleAttribute(value, float) @staticmethod - def _makeBoolAttribute(value): + def _makeBoolAttribute(value: Optional[bool]) -> Attribute[bool]: return GithubObject.__makeSimpleAttribute(value, bool) @staticmethod - def _makeDictAttribute(value): + def _makeDictAttribute(value: Dict[str, Any]) -> Attribute[dict]: return GithubObject.__makeSimpleAttribute(value, dict) @staticmethod - def _makeTimestampAttribute(value): + def _makeTimestampAttribute(value: int) -> Attribute[int]: return GithubObject.__makeTransformedAttribute( value, int, datetime.datetime.utcfromtimestamp ) @staticmethod - def _makeDatetimeAttribute(value): + def _makeDatetimeAttribute(value: Optional[Union[int, str]]) -> Attribute: def parseDatetime(s): if ( len(s) == 24 @@ -202,7 +248,7 @@ def parseDatetime(s): return GithubObject.__makeTransformedAttribute(value, str, parseDatetime) - def _makeClassAttribute(self, klass, value): + def _makeClassAttribute(self, klass: Any, value: Any) -> Attribute: return GithubObject.__makeTransformedAttribute( value, dict, @@ -210,22 +256,30 @@ def _makeClassAttribute(self, klass, value): ) @staticmethod - def _makeListOfStringsAttribute(value): + def _makeListOfStringsAttribute( + value: Union[List[List[str]], List[str], List[Union[str, int]]] + ) -> Attribute: return GithubObject.__makeSimpleListAttribute(value, str) @staticmethod - def _makeListOfIntsAttribute(value): + def _makeListOfIntsAttribute(value: List[int]) -> Attribute: return GithubObject.__makeSimpleListAttribute(value, int) @staticmethod - def _makeListOfDictsAttribute(value): + def _makeListOfDictsAttribute( + value: List[Dict[str, Union[str, List[Dict[str, Union[str, List[int]]]]]]] + ) -> Attribute: return GithubObject.__makeSimpleListAttribute(value, dict) @staticmethod - def _makeListOfListOfStringsAttribute(value): + def _makeListOfListOfStringsAttribute( + value: List[List[str]], + ) -> Attribute: return GithubObject.__makeSimpleListAttribute(value, list) - def _makeListOfClassesAttribute(self, klass, value): + def _makeListOfClassesAttribute( + self, klass: Any, value: Any + ) -> Union[_ValuedAttribute, _BadAttribute]: if isinstance(value, list) and all( isinstance(element, dict) for element in value ): @@ -238,7 +292,14 @@ def _makeListOfClassesAttribute(self, klass, value): else: return _BadAttribute(value, [dict]) - def _makeDictOfStringsToClassesAttribute(self, klass, value): + def _makeDictOfStringsToClassesAttribute( + self, + klass: Type["NonCompletableGithubObject"], + value: Dict[ + str, + Union[int, Dict[str, Union[str, int, None]], Dict[str, Union[str, int]]], + ], + ) -> Union[_ValuedAttribute, _BadAttribute]: if isinstance(value, dict) and all( isinstance(key, str) and isinstance(element, dict) for key, element in value.items() @@ -253,20 +314,20 @@ def _makeDictOfStringsToClassesAttribute(self, klass, value): return _BadAttribute(value, {str: dict}) @property - def etag(self): + def etag(self) -> Optional[str]: """ :type: str """ - return self._headers.get(Consts.RES_ETAG) + return self._headers.get(Consts.RES_ETAG) # type: ignore @property - def last_modified(self): + def last_modified(self) -> Optional[str]: """ :type: str """ - return self._headers.get(Consts.RES_LAST_MODIFIED) + return self._headers.get(Consts.RES_LAST_MODIFIED) # type: ignore - def get__repr__(self, params): + def get__repr__(self, params: Dict[str, Any]) -> str: """ Converts the object to a nicely printable string. """ @@ -285,44 +346,57 @@ def format_params(params): params=", ".join(list(format_params(params))), ) + def _initAttributes(self): + raise NotImplementedError("BUG: Not Implemented _initAttributes") + + def _useAttributes(self, attributes): + raise NotImplementedError("BUG: Not Implemented _useAttributes") -class NonCompletableGithubObject(GithubObject): def _completeIfNeeded(self): + raise NotImplementedError("BUG: Not Implemented _completeIfNeeded") + + +class NonCompletableGithubObject(GithubObject): + def _completeIfNeeded(self) -> None: pass class CompletableGithubObject(GithubObject): - def __init__(self, requester, headers, attributes, completed): + def __init__( + self, + requester: "Requester", + headers: Dict[str, Union[str, int]], + attributes: Dict[str, Any], + completed: bool, + ): super().__init__(requester, headers, attributes, completed) self.__completed = completed - def __eq__(self, other): + def __eq__(self, other: Any) -> bool: return other.__class__ is self.__class__ and other._url.value == self._url.value def __hash__(self): return hash(self._url.value) - def __ne__(self, other): + def __ne__(self, other: Any) -> bool: return not self == other - def _completeIfNotSet(self, value): - if value is NotSet: + def _completeIfNotSet(self, value: Attribute) -> None: + if isinstance(value, _NotSetType): self._completeIfNeeded() - def _completeIfNeeded(self): + def _completeIfNeeded(self) -> None: if not self.__completed: self.__complete() def __complete(self): if self._url.value is None: - raise GithubException.IncompletableObject( - 400, "Returned object contains no URL", None - ) + raise IncompletableObject(400, "Returned object contains no URL", None) headers, data = self._requester.requestJsonAndCheck("GET", self._url.value) self._storeAndUseAttributes(headers, data) self.__completed = True - def update(self, additional_headers=None): + def update(self, additional_headers: Optional[Dict[str, Any]] = None) -> bool: """ Check and update the object with conditional request :rtype: Boolean value indicating whether the object is changed @@ -341,7 +415,7 @@ def update(self, additional_headers=None): if status == 304: return False else: - headers, data = self._requester._Requester__check( + headers, data = self._requester._Requester__check( # type: ignore status, responseHeaders, output ) self._storeAndUseAttributes(headers, data) diff --git a/github/GithubObject.pyi b/github/GithubObject.pyi deleted file mode 100644 index cada21e852..0000000000 --- a/github/GithubObject.pyi +++ /dev/null @@ -1,134 +0,0 @@ -from typing import Any, Callable, Dict, List, Optional, Type, Union, TypeVar - -from github.Commit import Commit -from github.GistFile import GistFile -from github.GitRelease import GitRelease -from github.NamedUser import NamedUser -from github.Organization import Organization -from github.PullRequestReview import PullRequestReview -from github.Requester import Requester - -class GithubObject: - _requester: Optional[Requester] - def __init__( - self, - requester: Optional[Requester], - headers: Dict[str, Union[str, int]], - attributes: Any, - completed: bool, - ) -> None: ... - @staticmethod - def _makeBoolAttribute(value: Optional[bool]) -> _ValuedAttribute: ... - def _makeClassAttribute( - self, klass: Any, value: Any - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeDatetimeAttribute( - value: Optional[Union[int, str]] - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeDictAttribute(value: Dict[str, Any]) -> _ValuedAttribute: ... - def _makeDictOfStringsToClassesAttribute( - self, - klass: Type[GistFile], - value: Dict[ - str, - Union[int, Dict[str, Union[str, int, None]], Dict[str, Union[str, int]]], - ], - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeIntAttribute( - value: Optional[Union[int, str]] - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - def _makeListOfClassesAttribute( - self, klass: Any, value: Any - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeListOfDictsAttribute( - value: List[Dict[str, Union[str, List[Dict[str, Union[str, List[int]]]]]]] - ) -> _ValuedAttribute: ... - @staticmethod - def _makeListOfIntsAttribute(value: List[int]) -> _ValuedAttribute: ... - @staticmethod - def _makeListOfListOfStringsAttribute( - value: List[List[str]], - ) -> _ValuedAttribute: ... - @staticmethod - def _makeListOfStringsAttribute( - value: Union[List[List[str]], List[str], List[Union[str, int]]] - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def __makeSimpleAttribute( - value: Any, type: type - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def __makeSimpleListAttribute( - value: list, type: type - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def __makeTransformedAttribute( - value: Any, type: type, transform: Callable[..., Any] - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeStringAttribute( - value: Optional[Union[int, str]] - ) -> Union[_ValuedAttribute, _BadAttribute]: ... - @staticmethod - def _makeTimestampAttribute(value: int) -> _ValuedAttribute: ... - @staticmethod - def _parentUrl(url: str) -> str: ... - def _storeAndUseAttributes( - self, headers: Dict[str, Union[str, int]], attributes: Any - ) -> None: ... - @property - def etag(self) -> Optional[str]: ... - def get__repr__(self, params: Dict[str, Any]) -> str: ... - @property - def last_modified(self) -> Optional[str]: ... - @property - def raw_data(self) -> Dict[str, Any]: ... - @property - def raw_headers(self) -> Dict[str, Union[str, int]]: ... - @classmethod - def setCheckAfterInitFlag(cls, flag: bool) -> None: ... - -class NonCompletableGithubObject(GithubObject): - def _completeIfNeeded(self) -> None: ... - -class CompletableGithubObject(GithubObject): - _requester: Requester - def __eq__(self, other: Any) -> bool: ... - def __init__( - self, - requester: Requester, - headers: Dict[str, Union[str, int]], - attributes: Dict[str, Any], - completed: bool, - ) -> None: ... - def __ne__(self, other: Any) -> bool: ... - def _completeIfNeeded(self) -> None: ... - def _completeIfNotSet( - self, value: Union[_ValuedAttribute, _BadAttribute, _NotSetType] - ) -> None: ... - def update(self) -> bool: ... - -class _BadAttribute: - def __init__( - self, value: Any, expectedType: Any, exception: Optional[ValueError] = ... - ) -> None: ... - @property - def value(self) -> Any: ... - -class _NotSetType: - def __repr__(self) -> str: ... - @property - def value(self) -> Any: ... - -NotSet: _NotSetType - -T = TypeVar("T") - -OptionallySet = Union[T, _NotSetType] - -class _ValuedAttribute: - def __init__(self, value: Any) -> None: ... diff --git a/github/RepositoryAdvisory.py b/github/RepositoryAdvisory.py index 84a563b9b0..459ff493e7 100644 --- a/github/RepositoryAdvisory.py +++ b/github/RepositoryAdvisory.py @@ -326,24 +326,22 @@ def clear_credits(self): def edit( self, - summary: github.GithubObject.OptionallySet[str] = github.GithubObject.NotSet, - description: github.GithubObject.OptionallySet[ + summary: github.GithubObject.Opt[str] = github.GithubObject.NotSet, + description: github.GithubObject.Opt[str] = github.GithubObject.NotSet, + severity_or_cvss_vector_string: github.GithubObject.Opt[ str ] = github.GithubObject.NotSet, - severity_or_cvss_vector_string: github.GithubObject.OptionallySet[ - str - ] = github.GithubObject.NotSet, - cve_id: github.GithubObject.OptionallySet[str] = github.GithubObject.NotSet, - vulnerabilities: github.GithubObject.OptionallySet[ + cve_id: github.GithubObject.Opt[str] = github.GithubObject.NotSet, + vulnerabilities: github.GithubObject.Opt[ typing.Iterable[AdvisoryVulnerability] ] = github.GithubObject.NotSet, - cwe_ids: github.GithubObject.OptionallySet[ + cwe_ids: github.GithubObject.Opt[ typing.Iterable[str] ] = github.GithubObject.NotSet, - credits: github.GithubObject.OptionallySet[ + credits: github.GithubObject.Opt[ typing.Iterable[Credit] ] = github.GithubObject.NotSet, - state: github.GithubObject.OptionallySet[str] = github.GithubObject.NotSet, + state: github.GithubObject.Opt[str] = github.GithubObject.NotSet, ) -> "RepositoryAdvisory": """ :calls: `PATCH /repos/{owner}/{repo}/security-advisories/:advisory_id `_ diff --git a/github/Requester.py b/github/Requester.py index b6ce0beb42..0f1de7229e 100644 --- a/github/Requester.py +++ b/github/Requester.py @@ -50,6 +50,7 @@ # # ################################################################################ +import io import json import logging import mimetypes @@ -57,39 +58,69 @@ import re import time import urllib +import urllib.parse from io import IOBase -from typing import Generic, Optional, TypeVar +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Dict, + Generic, + ItemsView, + List, + Optional, + Tuple, + Type, + TypeVar, + Union, +) import requests +import requests.adapters +from urllib3 import Retry -from . import Consts, GithubException +import github.Consts as Consts +import github.GithubException as GithubException + +if TYPE_CHECKING: + from .AppAuthentication import AppAuthentication + from .Auth import Auth + from .GithubObject import GithubObject + from .InstallationAuthorization import InstallationAuthorization + +T = TypeVar("T") + +# For App authentication, time remaining before token expiration to request a new one +ACCESS_TOKEN_REFRESH_THRESHOLD_SECONDS = 20 class RequestsResponse: # mimic the httplib response object - def __init__(self, r): + def __init__(self, r: requests.Response): self.status = r.status_code self.headers = r.headers self.text = r.text - def getheaders(self): + def getheaders(self) -> ItemsView[str, str]: return self.headers.items() - def read(self): + def read(self) -> str: return self.text class HTTPSRequestsConnectionClass: + retry: Union[int, Retry] + # mimic the httplib connection object def __init__( self, host, - port=None, - strict=False, - timeout=None, - retry=None, - pool_size=None, - **kwargs, + port: Optional[int] = None, + strict: bool = False, + timeout: Optional[int] = None, + retry: Optional[Union[int, Retry]] = None, + pool_size: Optional[int] = None, + **kwargs: Any, ): self.port = port if port else 443 self.host = host @@ -115,13 +146,19 @@ def __init__( ) self.session.mount("https://", self.adapter) - def request(self, verb, url, input, headers): + def request( + self, + verb: str, + url: str, + input: Optional[Union[str, io.BufferedReader]], + headers: Dict[str, str], + ): self.verb = verb self.url = url self.input = input self.headers = headers - def getresponse(self): + def getresponse(self) -> RequestsResponse: verb = getattr(self.session, self.verb.lower()) url = f"{self.protocol}://{self.host}:{self.port}{self.url}" r = verb( @@ -142,13 +179,13 @@ class HTTPRequestsConnectionClass: # mimic the httplib connection object def __init__( self, - host, - port=None, - strict=False, - timeout=None, - retry=None, - pool_size=None, - **kwargs, + host: str, + port: Optional[int] = None, + strict: bool = False, + timeout: Optional[int] = None, + retry: Optional[Union[int, Retry]] = None, + pool_size: Optional[int] = None, + **kwargs: Any, ): self.port = port if port else 80 self.host = host @@ -160,7 +197,7 @@ def __init__( if retry is None: self.retry = requests.adapters.DEFAULT_RETRIES else: - self.retry = retry + self.retry = retry # type: ignore if pool_size is None: self.pool_size = requests.adapters.DEFAULT_POOLSIZE @@ -174,13 +211,13 @@ def __init__( ) self.session.mount("http://", self.adapter) - def request(self, verb, url, input, headers): + def request(self, verb: str, url: str, input: None, headers: Dict[str, str]): self.verb = verb self.url = url self.input = input self.headers = headers - def getresponse(self): + def getresponse(self) -> RequestsResponse: verb = getattr(self.session, self.verb.lower()) url = f"{self.protocol}://{self.host}:{self.port}{self.url}" r = verb( @@ -193,25 +230,34 @@ def getresponse(self): ) return RequestsResponse(r) - def close(self): + def close(self) -> None: return class Requester: + __installation_authorization: Optional["InstallationAuthorization"] + __app_auth: Optional["AppAuthentication"] + __httpConnectionClass = HTTPRequestsConnectionClass __httpsConnectionClass = HTTPSRequestsConnectionClass __connection = None __persist = True __logger = None + _frameBuffer: List[Any] + @classmethod - def injectConnectionClasses(cls, httpConnectionClass, httpsConnectionClass): + def injectConnectionClasses( + cls, + httpConnectionClass: Type[HTTPRequestsConnectionClass], + httpsConnectionClass: Type[HTTPSRequestsConnectionClass], + ): cls.__persist = False cls.__httpConnectionClass = httpConnectionClass cls.__httpsConnectionClass = httpsConnectionClass @classmethod - def resetConnectionClasses(cls): + def resetConnectionClasses(cls) -> None: cls.__persist = True cls.__httpConnectionClass = HTTPRequestsConnectionClass cls.__httpsConnectionClass = HTTPSRequestsConnectionClass @@ -227,11 +273,11 @@ def resetLogger(cls): ############################################################# # For Debug @classmethod - def setDebugFlag(cls, flag): + def setDebugFlag(cls, flag: bool) -> None: cls.DEBUG_FLAG = flag @classmethod - def setOnCheckMe(cls, onCheckMe): + def setOnCheckMe(cls, onCheckMe: Callable) -> None: cls.ON_CHECK_ME = onCheckMe DEBUG_FLAG = False @@ -240,9 +286,9 @@ def setOnCheckMe(cls, onCheckMe): DEBUG_HEADER_KEY = "DEBUG_FRAME" - ON_CHECK_ME = None + ON_CHECK_ME: Optional[Callable] = None - def NEW_DEBUG_FRAME(self, requestHeader): + def NEW_DEBUG_FRAME(self, requestHeader: Dict[str, str]) -> None: """ Initialize a debug frame with requestHeader Frame count is updated and will be attached to respond header @@ -260,7 +306,9 @@ def NEW_DEBUG_FRAME(self, requestHeader): self._frameCount = len(self._frameBuffer) - 1 - def DEBUG_ON_RESPONSE(self, statusCode, responseHeader, data): + def DEBUG_ON_RESPONSE( + self, statusCode: int, responseHeader: Dict[str, Union[str, int]], data: str + ): """ Update current frame with response Current frame index will be attached to responseHeader @@ -273,14 +321,14 @@ def DEBUG_ON_RESPONSE(self, statusCode, responseHeader, data): ] responseHeader[self.DEBUG_HEADER_KEY] = self._frameCount - def check_me(self, obj): + def check_me(self, obj: "GithubObject"): if ( self.DEBUG_FLAG and self.ON_CHECK_ME is not None ): # pragma no branch (Flag always set in tests) frame = None if self.DEBUG_HEADER_KEY in obj._headers: frame_index = obj._headers[self.DEBUG_HEADER_KEY] - frame = self._frameBuffer[frame_index] + frame = self._frameBuffer[frame_index] # type: ignore self.ON_CHECK_ME(obj, frame) def _initializeDebugFeature(self): @@ -289,16 +337,23 @@ def _initializeDebugFeature(self): ############################################################# + _frameCount: int + __connectionClass: Union[ + Type[HTTPRequestsConnectionClass], Type[HTTPSRequestsConnectionClass] + ] + __hostname: str + __authorizationHeader: Optional[str] + def __init__( self, - auth, - base_url, - timeout, - user_agent, - per_page, - verify, - retry, - pool_size, + auth: Optional["Auth"], + base_url: str, + timeout: int, + user_agent: str, + per_page: int, + verify: bool, + retry: Optional[Union[int, Retry]], + pool_size: Optional[int], ): self._initializeDebugFeature() @@ -306,7 +361,7 @@ def __init__( self.__base_url = base_url o = urllib.parse.urlparse(base_url) - self.__hostname = o.hostname + self.__hostname = o.hostname # type: ignore self.__port = o.port self.__prefix = o.path self.__timeout = timeout @@ -340,14 +395,14 @@ def __init__( self.__auth.withRequester(self) @property - def base_url(self): + def base_url(self) -> str: return self.__base_url @property - def auth(self): + def auth(self) -> Optional["Auth"]: return self.__auth - def withAuth(self, auth): + def withAuth(self, auth: Optional["Auth"]) -> "Requester": """ Create a new requester instance with identical configuration but the given authentication method. :param auth: authentication method @@ -364,7 +419,14 @@ def withAuth(self, auth): pool_size=self.__pool_size, ) - def requestJsonAndCheck(self, verb, url, parameters=None, headers=None, input=None): + def requestJsonAndCheck( + self, + verb: str, + url: str, + parameters: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + input: Optional[Any] = None, + ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: return self.__check( *self.requestJson( verb, url, parameters, headers, input, self.__customConnection(url) @@ -372,29 +434,53 @@ def requestJsonAndCheck(self, verb, url, parameters=None, headers=None, input=No ) def requestMultipartAndCheck( - self, verb, url, parameters=None, headers=None, input=None - ): + self, + verb: str, + url: str, + parameters: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, Any]] = None, + input: Optional[Dict[str, str]] = None, + ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: return self.__check( *self.requestMultipart( verb, url, parameters, headers, input, self.__customConnection(url) ) ) - def requestBlobAndCheck(self, verb, url, parameters=None, headers=None, input=None): + def requestBlobAndCheck( + self, + verb: str, + url: str, + parameters: Optional[Dict[str, str]] = None, + headers: Optional[Dict[str, str]] = None, + input: Optional[str] = None, + cnx: Optional[ + Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] + ] = None, + ) -> Tuple[Dict[str, Any], Dict[str, Any]]: return self.__check( *self.requestBlob( verb, url, parameters, headers, input, self.__customConnection(url) ) ) - def __check(self, status, responseHeaders, output): - output = self.__structuredFromJson(output) + def __check( + self, + status: int, + responseHeaders: Dict[str, Any], + output: str, + ) -> Tuple[Dict[str, Any], Dict[str, Any]]: + data = self.__structuredFromJson(output) if status >= 400: - raise self.__createException(status, responseHeaders, output) - return responseHeaders, output - - def __customConnection(self, url): - cnx = None + raise self.__createException(status, responseHeaders, data) + return responseHeaders, data + + def __customConnection( + self, url: str + ) -> Optional[Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]]: + cnx: Optional[ + Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] + ] = None if not url.startswith("/"): o = urllib.parse.urlparse(url) if ( @@ -407,7 +493,7 @@ def __customConnection(self, url): ): # issue80 if o.scheme == "http": cnx = self.__httpConnectionClass( - o.hostname, + o.hostname, # type: ignore o.port, retry=self.__retry, pool_size=self.__pool_size, @@ -421,9 +507,15 @@ def __customConnection(self, url): ) return cnx - def __createException(self, status, headers, output): + def __createException( + self, + status: int, + headers: Dict[str, Any], + output: Dict[str, Any], + ) -> Any: message = output.get("message", "").lower() if output is not None else "" + cls = GithubException.GithubException if status == 401 and message == "bad credentials": cls = GithubException.BadCredentialsException elif ( @@ -443,11 +535,10 @@ def __createException(self, status, headers, output): cls = GithubException.RateLimitExceededException elif status == 404 and message == "not found": cls = GithubException.UnknownObjectException - else: - cls = GithubException.GithubException + return cls(status, output, headers) - def __structuredFromJson(self, data): + def __structuredFromJson(self, data: str) -> Any: if len(data) == 0: return None else: @@ -461,16 +552,32 @@ def __structuredFromJson(self, data): return {"data": data} def requestJson( - self, verb, url, parameters=None, headers=None, input=None, cnx=None - ): + self, + verb: str, + url: str, + parameters: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, Any]] = None, + input: Optional[Any] = None, + cnx: Optional[ + Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] + ] = None, + ) -> Tuple[int, Dict[str, Any], str]: def encode(input): return "application/json", json.dumps(input) return self.__requestEncode(cnx, verb, url, parameters, headers, input, encode) def requestMultipart( - self, verb, url, parameters=None, headers=None, input=None, cnx=None - ): + self, + verb: str, + url: str, + parameters: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, Any]] = None, + input: Optional[Dict[str, str]] = None, + cnx: Optional[ + Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] + ] = None, + ) -> Tuple[int, Dict[str, Any], str]: def encode(input): boundary = "----------------------------3c3ba8b523b2" eol = "\r\n" @@ -486,12 +593,25 @@ def encode(input): return self.__requestEncode(cnx, verb, url, parameters, headers, input, encode) - def requestBlob(self, verb, url, parameters={}, headers={}, input=None, cnx=None): - def encode(local_path): - if "Content-Type" in headers: - mime_type = headers["Content-Type"] + def requestBlob( + self, + verb: str, + url: str, + parameters: Optional[Dict[str, str]] = None, + headers: Optional[Dict[str, str]] = None, + input: Optional[str] = None, + cnx: Optional[ + Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] + ] = None, + ) -> Tuple[int, Dict[str, Any], str]: + if headers is None: + headers = {} + + def encode(local_path: str): + if "Content-Type" in headers: # type: ignore + mime_type = headers["Content-Type"] # type: ignore else: - guessed_type = mimetypes.guess_type(input) + guessed_type = mimetypes.guess_type(local_path) mime_type = ( guessed_type[0] if guessed_type[0] is not None @@ -520,13 +640,20 @@ def encode(_): ) def __requestEncode( - self, cnx, verb, url, parameters, requestHeaders, input, encode - ): + self, + cnx: Optional[Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]], + verb: str, + url: str, + parameters: Optional[Dict[str, str]], + requestHeaders: Optional[Dict[str, str]], + input: Optional[T], + encode: Callable[[T], Tuple[str, Any]], + ) -> Tuple[int, Dict[str, Any], str]: assert verb in ["HEAD", "GET", "POST", "PATCH", "PUT", "DELETE"] if parameters is None: - parameters = dict() + parameters = {} if requestHeaders is None: - requestHeaders = dict() + requestHeaders = {} if self.__auth is not None: requestHeaders[ @@ -565,7 +692,14 @@ def __requestEncode( return status, responseHeaders, output - def __requestRaw(self, cnx, verb, url, requestHeaders, input): + def __requestRaw( + self, + cnx: Optional[Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]], + verb: str, + url: str, + requestHeaders: Dict[str, str], + input: Optional[Any], + ) -> Tuple[int, Dict[str, Any], str]: original_cnx = cnx if cnx is None: cnx = self.__createConnection() @@ -618,7 +752,7 @@ def __requestRaw(self, cnx, verb, url, requestHeaders, input): return status, responseHeaders, output - def __makeAbsoluteUrl(self, url): + def __makeAbsoluteUrl(self, url: str) -> str: # URLs generated locally will be relative to __base_url # URLs returned from the server will start with __base_url if url.startswith("/"): @@ -638,17 +772,19 @@ def __makeAbsoluteUrl(self, url): url += f"?{o.query}" return url - def __addParametersToUrl(self, url, parameters): + def __addParametersToUrl( + self, + url: str, + parameters: Dict[str, Any], + ): if len(parameters) == 0: return url else: return f"{url}?{urllib.parse.urlencode(parameters)}" - def __createConnection(self): - kwds = {} - kwds["timeout"] = self.__timeout - kwds["verify"] = self.__verify - + def __createConnection( + self, + ) -> Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]: if self.__persist and self.__connection is not None: return self.__connection @@ -657,18 +793,28 @@ def __createConnection(self): self.__port, retry=self.__retry, pool_size=self.__pool_size, - **kwds, + timeout=self.__timeout, + verify=self.__verify, ) return self.__connection @property - def _logger(self): + def _logger(self) -> logging.Logger: if self.__logger is None: self.__logger = logging.getLogger(__name__) return self.__logger - def __log(self, verb, url, requestHeaders, input, status, responseHeaders, output): + def __log( + self, + verb: str, + url: str, + requestHeaders: Dict[str, str], + input: Optional[Any], + status: Optional[int], + responseHeaders: Dict[str, Any], + output: Optional[str], + ) -> None: if self._logger.isEnabledFor(logging.DEBUG): headersForRequest = requestHeaders.copy() if "Authorization" in requestHeaders: @@ -698,14 +844,13 @@ def __log(self, verb, url, requestHeaders, input, status, responseHeaders, outpu ) -T = TypeVar("T") - - class WithRequester(Generic[T]): """ Mixin class that allows to set a requester. """ + __requester: Requester + def __init__(self): self.__requester: Optional[Requester] = None @@ -713,7 +858,7 @@ def __init__(self): def requester(self) -> Requester: return self.__requester - def withRequester(self, requester: Requester) -> T: + def withRequester(self, requester: Requester) -> "WithRequester[T]": assert isinstance(requester, Requester), requester self.__requester = requester return self diff --git a/github/Requester.pyi b/github/Requester.pyi deleted file mode 100644 index 9d381816de..0000000000 --- a/github/Requester.pyi +++ /dev/null @@ -1,232 +0,0 @@ -from collections import OrderedDict -from io import BufferedReader -from logging import Logger -from typing import ( - Any, - Callable, - Dict, - Iterator, - Optional, - Tuple, - Union, - Generic, - TypeVar, -) - -from requests.models import Response -from urllib3.util import Retry - -from github.Auth import Auth -from github.GithubObject import GithubObject -from github.InstallationAuthorization import InstallationAuthorization - -class HTTPRequestsConnectionClass: - def __init__( - self, - host: str, - port: Optional[int] = ..., - strict: bool = ..., - timeout: Optional[int] = ..., - retry: Optional[Union[int, Retry]] = ..., - pool_size: Optional[int] = ..., - **kwargs: str - ) -> None: ... - def close(self) -> None: ... - def getresponse(self) -> RequestsResponse: ... - def request( - self, verb: str, url: str, input: None, headers: Dict[str, str] - ) -> None: ... - -class HTTPSRequestsConnectionClass: - def __init__( - self, - host: str, - port: Optional[int] = ..., - strict: bool = ..., - timeout: Optional[int] = ..., - retry: Optional[Union[int, Retry]] = ..., - pool_size: Optional[int] = ..., - **kwargs: str - ) -> None: ... - def close(self) -> None: ... - def getresponse(self) -> RequestsResponse: ... - def request( - self, - verb: str, - url: str, - input: Optional[Union[str, BufferedReader]], - headers: Dict[str, str], - ) -> None: ... - -class Requester: - __auth: Optional[Auth] = ... - __installation_authorization: Optional[InstallationAuthorization] = ... - __logger: Logger - def DEBUG_ON_RESPONSE( - self, statusCode: int, responseHeader: Dict[str, str], data: str - ) -> None: ... - def NEW_DEBUG_FRAME(self, requestHeader: Dict[str, str]) -> None: ... - def __check( - self, - status: int, - responseHeaders: Dict[str, Any], - output: str, - ) -> Tuple[Dict[str, Any], Dict[str, Any]]: ... - def __addParametersToUrl( - self, - url: str, - parameters: Dict[str, Any], - ) -> str: ... - def __authenticate( - self, - url: str, - requestHeaders: Dict[str, Any], - parameters: Dict[str, Any], - ) -> None: ... - def __customConnection( - self, - url: str, - ) -> Optional[Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]]: ... - def __createConnection( - self, - ) -> Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass]: ... - def __createException( - self, - status: int, - headers: Dict[str, Any], - output: str, - ) -> Any: ... - @property - def _logger(self) -> Logger: ... - def __log( - self, - verb: str, - url: str, - requestHeaders: Dict[str, str], - input: Optional[str], - status: Optional[int], - responseHeaders: Dict[str, Any], - output: Optional[str], - ) -> None: ... - def __makeAbsoluteUrl(self, url: str) -> str: ... - def __structuredFromJson(self, data: str) -> Optional[Dict[str, Any]]: ... - def __requestEncode( - self, - cnx: Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass], - verb: str, - url: str, - parameters: Dict[str, str], - requestHeaders: Dict[str, str], - input: Optional[str], - encode: Callable[[str], str], - ) -> Tuple[int, Dict[str, Any], str]: ... - def __requestRaw( - self, - cnx: Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass], - verb: str, - url: str, - requestHeaders: Dict[str, str], - input: Optional[str], - ) -> Tuple[int, Dict[str, Any], str]: ... - def __init__( - self, - auth: Optional[Auth], - base_url: str, - timeout: int, - user_agent: str, - per_page: int, - verify: bool, - retry: Optional[Union[int, Retry]], - pool_size: Optional[int], - ) -> None: ... - @property - def base_url(self) -> str: ... - @property - def auth(self) -> Optional[Auth]: ... - def withAuth(self, auth: Optional[Auth]) -> Requester: ... - def _initializeDebugFeature(self) -> None: ... - def check_me(self, obj: GithubObject) -> None: ... - def _must_refresh_token(self) -> bool: ... - def _get_installation_authorization(self) -> InstallationAuthorization: ... - def _refresh_token_if_needed(self) -> None: ... - def _refresh_token(self) -> None: ... - @classmethod - def injectConnectionClasses( - cls, httpConnectionClass: Callable, httpsConnectionClass: Callable - ) -> None: ... - def requestBlob( - self, - verb: str, - url: str, - parameters: Dict[str, str] = ..., - headers: Dict[str, str] = ..., - input: Optional[str] = ..., - cnx: Optional[ - Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] - ] = ..., - ) -> Tuple[int, Dict[str, Any], str]: ... - def requestBlobAndCheck( - self, - verb: str, - url: str, - parameters: Optional[Dict[str, Any]] = ..., - headers: Optional[Dict[str, Any]] = ..., - input: Optional[str] = ..., - ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: ... - def requestJson( - self, - verb: str, - url: str, - parameters: Optional[Dict[str, Any]] = ..., - headers: Optional[Dict[str, Any]] = ..., - input: Optional[Any] = ..., - cnx: Optional[ - Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] - ] = ..., - ) -> Tuple[int, Dict[str, Any], str]: ... - def requestJsonAndCheck( - self, - verb: str, - url: str, - parameters: Optional[Dict[str, Any]] = ..., - headers: Optional[Dict[str, str]] = ..., - input: Optional[Any] = ..., - ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: ... - def requestMultipart( - self, - verb: str, - url: str, - parameters: Optional[Dict[str, Any]] = ..., - headers: Optional[Dict[str, Any]] = ..., - input: Optional[OrderedDict[str, str]] = ..., - cnx: Optional[ - Union[HTTPRequestsConnectionClass, HTTPSRequestsConnectionClass] - ] = ..., - ) -> Tuple[int, Dict[str, Any], str]: ... - def requestMultipartAndCheck( - self, - verb: str, - url: str, - parameters: Optional[Dict[str, Any]] = ..., - headers: Optional[Dict[str, Any]] = ..., - input: Optional[OrderedDict[str, str]] = ..., - ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: ... - @classmethod - def resetConnectionClasses(cls) -> None: ... - @classmethod - def setDebugFlag(cls, flag: bool) -> None: ... - @classmethod - def setOnCheckMe(cls, onCheckMe: Callable) -> None: ... - -T = TypeVar("T") - -class WithRequester(Generic[T]): - __requester: Optional[Requester] - @property - def requester(self) -> Requester: ... - def withRequester(self, requester: Requester) -> T: ... - -class RequestsResponse: - def __init__(self, r: Response) -> None: ... - def getheaders(self) -> Iterator[Any]: ... - def read(self) -> str: ...