merge GithubObject.pyi/Requester.pyi stubs back to source (#2463)

Co-authored-by: Enrico Minack <github@enrico.minack.dev>
Co-authored-by: Jonathan Leitschuh <jonathan.leitschuh@gmail.com>
This commit is contained in:
Trim21
2023-06-16 11:57:35 -04:00
committed by GitHub
co-authored by Enrico Minack Jonathan Leitschuh
parent 6d4b6d1419
commit b6258f4be9
10 changed files with 461 additions and 680 deletions
+24 -19
View File
@@ -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 @@ class AccessToken(github.GithubObject.NonCompletableGithubObject):
)
@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 @@ class AccessToken(github.GithubObject.NonCompletableGithubObject):
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]
"""
-29
View File
@@ -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: ...
+58 -69
View File
@@ -27,9 +27,15 @@
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
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 @@ class Authorization(github.GithubObject.CompletableGithubObject):
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} <https://docs.github.com/en/developers/apps/authorizing-oauth-apps>`_
: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} <https://docs.github.com/en/developers/apps/authorizing-oauth-apps>`_
:param scopes: list of string
@@ -136,45 +127,43 @@ class Authorization(github.GithubObject.CompletableGithubObject):
: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
-37
View File
@@ -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: ...
+4 -2
View File
@@ -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 @@ class GithubException(Exception):
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
"""
+134 -60
View File
@@ -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, GithubException
from . import Consts
from .GithubException import BadAttributeException, IncompletableObject
if TYPE_CHECKING:
from .Requester import Requester
T = typing.TypeVar("T")
class _NotSetType:
class Attribute(Generic[T]):
@property
def value(self) -> T:
raise NotImplementedError
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")
OptionallySet = typing.Union[T, _NotSetType]
Opt = Union[T, _NotSetType]
class _ValuedAttribute:
def __init__(self, value):
self.value = value
class _ValuedAttribute(Attribute, Generic[T]):
def __init__(self, value: T):
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 @@ class GithubObject:
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 @@ class GithubObject:
self._useAttributes(attributes)
@property
def raw_data(self):
def raw_data(self) -> Dict[str, Any]:
"""
:type: dict
"""
@@ -117,7 +161,7 @@ class GithubObject:
return self._rawData
@property
def raw_headers(self):
def raw_headers(self) -> Dict[str, Union[str, int]]:
"""
:type: dict
"""
@@ -125,43 +169,45 @@ class GithubObject:
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 @@ class GithubObject:
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 @@ class GithubObject:
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 @@ class GithubObject:
)
@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 @@ class GithubObject:
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 @@ class GithubObject:
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 @@ class GithubObject:
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")
def _completeIfNeeded(self):
raise NotImplementedError("BUG: Not Implemented _completeIfNeeded")
class NonCompletableGithubObject(GithubObject):
def _completeIfNeeded(self):
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 @@ class CompletableGithubObject(GithubObject):
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)
-134
View File
@@ -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: ...
+8 -10
View File
@@ -326,24 +326,22 @@ class RepositoryAdvisory(github.GithubObject.NonCompletableGithubObject):
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 <https://docs.github.com/en/rest/security-advisories/repository-advisories>`_
+233 -88
View File
@@ -50,6 +50,7 @@
# #
################################################################################
import io
import json
import logging
import mimetypes
@@ -57,39 +58,69 @@ import os
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 @@ class HTTPSRequestsConnectionClass:
)
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 @@ class HTTPRequestsConnectionClass:
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 @@ class HTTPRequestsConnectionClass:
)
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 @@ class HTTPRequestsConnectionClass:
)
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 @@ class Requester:
#############################################################
# 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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
]
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 @@ class Requester:
#############################################################
_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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
)
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
raise self.__createException(status, responseHeaders, data)
return responseHeaders, data
def __customConnection(self, url):
cnx = None
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 @@ class Requester:
): # 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 @@ class Requester:
)
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
)
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
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 @@ class Requester:
)
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 @@ class WithRequester(Generic[T]):
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
-232
View File
@@ -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: ...