diff --git a/github/AccessToken.py b/github/AccessToken.py
index 98169fa2..7cea4e86 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 @@ 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]
"""
diff --git a/github/AccessToken.pyi b/github/AccessToken.pyi
deleted file mode 100644
index a0f959f9..00000000
--- 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 b495d4b3..5fe99f86 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 @@ 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} `_
- :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 @@ 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
diff --git a/github/Authorization.pyi b/github/Authorization.pyi
deleted file mode 100644
index 059485ce..00000000
--- 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 46fb0414..717102ca 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 @@ 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
"""
diff --git a/github/GithubObject.py b/github/GithubObject.py
index 3ad6f867..6cea1cc0 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, 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)
diff --git a/github/GithubObject.pyi b/github/GithubObject.pyi
deleted file mode 100644
index cada21e8..00000000
--- 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 84a563b9..459ff493 100644
--- a/github/RepositoryAdvisory.py
+++ b/github/RepositoryAdvisory.py
@@ -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 `_
diff --git a/github/Requester.py b/github/Requester.py
index b6ce0beb..0f1de722 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 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
diff --git a/github/Requester.pyi b/github/Requester.pyi
deleted file mode 100644
index 9d381816..00000000
--- 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: ...