GithubObject.py
1 ############################ Copyrights and license ############################ 2 # # 3 # Copyright 2012 Vincent Jacques <vincent@vincent-jacques.net> # 4 # Copyright 2012 Zearin <zearin@gonk.net> # 5 # Copyright 2013 AKFish <akfish@gmail.com> # 6 # Copyright 2013 Vincent Jacques <vincent@vincent-jacques.net> # 7 # Copyright 2014 Andrew Scheller <github@loowis.durge.org> # 8 # Copyright 2014 Vincent Jacques <vincent@vincent-jacques.net> # 9 # Copyright 2016 Jakub Wilk <jwilk@jwilk.net> # 10 # Copyright 2016 Jannis Gebauer <ja.geb@me.com> # 11 # Copyright 2016 Peter Buckley <dx-pbuckley@users.noreply.github.com> # 12 # Copyright 2016 Sam Corbett <sam.corbett@cloudsoftcorp.com> # 13 # Copyright 2018 Shubham Singh <41840111+singh811@users.noreply.github.com> # 14 # Copyright 2018 h.shi <10385628+AnYeMoWang@users.noreply.github.com> # 15 # Copyright 2018 sfdye <tsfdye@gmail.com> # 16 # Copyright 2019 Adam Baratz <adam.baratz@gmail.com> # 17 # Copyright 2019 Steve Kowalik <steven@wedontsleep.org> # 18 # Copyright 2019 Wan Liuyang <tsfdye@gmail.com> # 19 # Copyright 2020 Steve Kowalik <steven@wedontsleep.org> # 20 # Copyright 2021 Steve Kowalik <steven@wedontsleep.org> # 21 # Copyright 2023 Jonathan Leitschuh <Jonathan.Leitschuh@gmail.com> # 22 # # 23 # This file is part of PyGithub. # 24 # http://pygithub.readthedocs.io/ # 25 # # 26 # PyGithub is free software: you can redistribute it and/or modify it under # 27 # the terms of the GNU Lesser General Public License as published by the Free # 28 # Software Foundation, either version 3 of the License, or (at your option) # 29 # any later version. # 30 # # 31 # PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY # 32 # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # 33 # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # 34 # details. # 35 # # 36 # You should have received a copy of the GNU Lesser General Public License # 37 # along with PyGithub. If not, see <http://www.gnu.org/licenses/>. # 38 # # 39 ################################################################################ 40 41 import typing 42 from datetime import datetime, timezone 43 from operator import itemgetter 44 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Type, Union 45 46 from dateutil import parser 47 from typing_extensions import Protocol, TypeGuard 48 49 from . import Consts 50 from .GithubException import BadAttributeException, IncompletableObject 51 52 if TYPE_CHECKING: 53 from .Requester import Requester 54 55 T = typing.TypeVar("T") 56 K = typing.TypeVar("K") 57 T_co = typing.TypeVar("T_co", covariant=True) 58 T_gh = typing.TypeVar("T_gh", bound="GithubObject") 59 60 61 class Attribute(Protocol[T_co]): 62 @property 63 def value(self) -> T_co: 64 raise NotImplementedError 65 66 67 class _NotSetType: 68 def __repr__(self) -> str: 69 return "NotSet" 70 71 @property 72 def value(self) -> Any: 73 return None 74 75 @staticmethod 76 def remove_unset_items(data: Dict[str, Any]) -> Dict[str, Any]: 77 return {key: value for key, value in data.items() if not isinstance(value, _NotSetType)} 78 79 80 NotSet = _NotSetType() 81 82 Opt = Union[T, _NotSetType] 83 84 85 def is_defined(v: Union[T, _NotSetType]) -> TypeGuard[T]: 86 return not isinstance(v, _NotSetType) 87 88 89 def is_undefined(v: Union[T, _NotSetType]) -> TypeGuard[_NotSetType]: 90 return isinstance(v, _NotSetType) 91 92 93 def is_optional(v: Any, type: Union[Type, Tuple[Type, ...]]) -> bool: 94 return isinstance(v, _NotSetType) or isinstance(v, type) 95 96 97 def is_optional_list(v: Any, type: Union[Type, Tuple[Type, ...]]) -> bool: 98 return isinstance(v, _NotSetType) or isinstance(v, list) and all(isinstance(element, type) for element in v) 99 100 101 class _ValuedAttribute(Attribute[T]): 102 def __init__(self, value: T): 103 self._value = value 104 105 @property 106 def value(self) -> T: 107 return self._value 108 109 110 class _BadAttribute(Attribute): 111 def __init__(self, value: Any, expectedType: Any, exception: Optional[Exception] = None): 112 self.__value = value 113 self.__expectedType = expectedType 114 self.__exception = exception 115 116 @property 117 def value(self) -> Any: 118 raise BadAttributeException(self.__value, self.__expectedType, self.__exception) 119 120 121 # v3: add * to edit function of all GithubObject implementations, 122 # this allows to rename attributes and maintain the order of attributes 123 class GithubObject: 124 """ 125 Base class for all classes representing objects returned by the API. 126 """ 127 128 """ 129 A global debug flag to enable header validation by requester for all objects 130 """ 131 CHECK_AFTER_INIT_FLAG = False 132 _url: Attribute[str] 133 134 @classmethod 135 def setCheckAfterInitFlag(cls, flag: bool) -> None: 136 cls.CHECK_AFTER_INIT_FLAG = flag 137 138 def __init__( 139 self, 140 requester: "Requester", 141 headers: Dict[str, Union[str, int]], 142 attributes: Any, 143 completed: bool, 144 ): 145 self._requester = requester 146 self._initAttributes() 147 self._storeAndUseAttributes(headers, attributes) 148 149 # Ask requester to do some checking, for debug and test purpose 150 # Since it's most handy to access and kinda all-knowing 151 if self.CHECK_AFTER_INIT_FLAG: # pragma no branch (Flag always set in tests) 152 requester.check_me(self) 153 154 def _storeAndUseAttributes(self, headers: Dict[str, Union[str, int]], attributes: Any) -> None: 155 # Make sure headers are assigned before calling _useAttributes 156 # (Some derived classes will use headers in _useAttributes) 157 self._headers = headers 158 self._rawData = attributes 159 self._useAttributes(attributes) 160 161 @property 162 def raw_data(self) -> Dict[str, Any]: 163 """ 164 :type: dict 165 """ 166 self._completeIfNeeded() 167 return self._rawData 168 169 @property 170 def raw_headers(self) -> Dict[str, Union[str, int]]: 171 """ 172 :type: dict 173 """ 174 self._completeIfNeeded() 175 return self._headers 176 177 @staticmethod 178 def _parentUrl(url: str) -> str: 179 return "/".join(url.split("/")[:-1]) 180 181 @staticmethod 182 def __makeSimpleAttribute(value: Any, type: Type[T]) -> Attribute[T]: 183 if value is None or isinstance(value, type): 184 return _ValuedAttribute(value) # type: ignore 185 else: 186 return _BadAttribute(value, type) # type: ignore 187 188 @staticmethod 189 def __makeSimpleListAttribute(value: list, type: Type[T]) -> Attribute[T]: 190 if isinstance(value, list) and all(isinstance(element, type) for element in value): 191 return _ValuedAttribute(value) # type: ignore 192 else: 193 return _BadAttribute(value, [type]) # type: ignore 194 195 @staticmethod 196 def __makeTransformedAttribute(value: T, type: Type[T], transform: Callable[[T], K]) -> Attribute[K]: 197 if value is None: 198 return _ValuedAttribute(None) # type: ignore 199 elif isinstance(value, type): 200 try: 201 return _ValuedAttribute(transform(value)) 202 except Exception as e: 203 return _BadAttribute(value, type, e) # type: ignore 204 else: 205 return _BadAttribute(value, type) # type: ignore 206 207 @staticmethod 208 def _makeStringAttribute(value: Optional[Union[int, str]]) -> Attribute[str]: 209 return GithubObject.__makeSimpleAttribute(value, str) 210 211 @staticmethod 212 def _makeIntAttribute(value: Optional[Union[int, str]]) -> Attribute[int]: 213 return GithubObject.__makeSimpleAttribute(value, int) 214 215 @staticmethod 216 def _makeFloatAttribute(value: Optional[float]) -> Attribute[float]: 217 return GithubObject.__makeSimpleAttribute(value, float) 218 219 @staticmethod 220 def _makeBoolAttribute(value: Optional[bool]) -> Attribute[bool]: 221 return GithubObject.__makeSimpleAttribute(value, bool) 222 223 @staticmethod 224 def _makeDictAttribute(value: Dict[str, Any]) -> Attribute[Dict[str, Any]]: 225 return GithubObject.__makeSimpleAttribute(value, dict) 226 227 @staticmethod 228 def _makeTimestampAttribute(value: int) -> Attribute[datetime]: 229 return GithubObject.__makeTransformedAttribute( 230 value, 231 int, 232 lambda t: datetime.fromtimestamp(t, tz=timezone.utc), 233 ) 234 235 @staticmethod 236 def _makeDatetimeAttribute(value: Optional[str]) -> Attribute[datetime]: 237 return GithubObject.__makeTransformedAttribute(value, str, parser.parse) # type: ignore 238 239 def _makeClassAttribute(self, klass: Type[T_gh], value: Any) -> Attribute[T_gh]: 240 return GithubObject.__makeTransformedAttribute( 241 value, 242 dict, 243 lambda value: klass(self._requester, self._headers, value, completed=False), 244 ) 245 246 @staticmethod 247 def _makeListOfStringsAttribute(value: Union[List[List[str]], List[str], List[Union[str, int]]]) -> Attribute: 248 return GithubObject.__makeSimpleListAttribute(value, str) 249 250 @staticmethod 251 def _makeListOfIntsAttribute(value: List[int]) -> Attribute: 252 return GithubObject.__makeSimpleListAttribute(value, int) 253 254 @staticmethod 255 def _makeListOfDictsAttribute( 256 value: List[Dict[str, Union[str, List[Dict[str, Union[str, List[int]]]]]]] 257 ) -> Attribute: 258 return GithubObject.__makeSimpleListAttribute(value, dict) 259 260 @staticmethod 261 def _makeListOfListOfStringsAttribute( 262 value: List[List[str]], 263 ) -> Attribute: 264 return GithubObject.__makeSimpleListAttribute(value, list) 265 266 def _makeListOfClassesAttribute(self, klass: Type[T_gh], value: Any) -> Attribute[List[T_gh]]: 267 if isinstance(value, list) and all(isinstance(element, dict) for element in value): 268 return _ValuedAttribute( 269 [klass(self._requester, self._headers, element, completed=False) for element in value] 270 ) 271 else: 272 return _BadAttribute(value, [dict]) 273 274 def _makeDictOfStringsToClassesAttribute( 275 self, 276 klass: Type[T_gh], 277 value: Dict[ 278 str, 279 Union[int, Dict[str, Union[str, int, None]], Dict[str, Union[str, int]]], 280 ], 281 ) -> Attribute[Dict[str, T_gh]]: 282 if isinstance(value, dict) and all( 283 isinstance(key, str) and isinstance(element, dict) for key, element in value.items() 284 ): 285 return _ValuedAttribute( 286 {key: klass(self._requester, self._headers, element, completed=False) for key, element in value.items()} 287 ) 288 else: 289 return _BadAttribute(value, {str: dict}) 290 291 @property 292 def etag(self) -> Optional[str]: 293 """ 294 :type: str 295 """ 296 return self._headers.get(Consts.RES_ETAG) # type: ignore 297 298 @property 299 def last_modified(self) -> Optional[str]: 300 """ 301 :type: str 302 """ 303 return self._headers.get(Consts.RES_LAST_MODIFIED) # type: ignore 304 305 def get__repr__(self, params: Dict[str, Any]) -> str: 306 """ 307 Converts the object to a nicely printable string. 308 """ 309 310 def format_params(params: Dict[str, Any]) -> typing.Generator[str, None, None]: 311 items = list(params.items()) 312 for k, v in sorted(items, key=itemgetter(0), reverse=True): 313 if isinstance(v, bytes): 314 v = v.decode("utf-8") 315 if isinstance(v, str): 316 v = f'"{v}"' 317 yield f"{k}={v}" 318 319 return "{class_name}({params})".format( 320 class_name=self.__class__.__name__, 321 params=", ".join(list(format_params(params))), 322 ) 323 324 def _initAttributes(self) -> None: 325 raise NotImplementedError("BUG: Not Implemented _initAttributes") 326 327 def _useAttributes(self, attributes: Any) -> None: 328 raise NotImplementedError("BUG: Not Implemented _useAttributes") 329 330 def _completeIfNeeded(self) -> None: 331 raise NotImplementedError("BUG: Not Implemented _completeIfNeeded") 332 333 334 class NonCompletableGithubObject(GithubObject): 335 def _completeIfNeeded(self) -> None: 336 pass 337 338 339 class CompletableGithubObject(GithubObject): 340 def __init__( 341 self, 342 requester: "Requester", 343 headers: Dict[str, Union[str, int]], 344 attributes: Dict[str, Any], 345 completed: bool, 346 ): 347 super().__init__(requester, headers, attributes, completed) 348 self.__completed = completed 349 350 def __eq__(self, other: Any) -> bool: 351 return other.__class__ is self.__class__ and other._url.value == self._url.value 352 353 def __hash__(self) -> int: 354 return hash(self._url.value) 355 356 def __ne__(self, other: Any) -> bool: 357 return not self == other 358 359 def _completeIfNotSet(self, value: Attribute) -> None: 360 if isinstance(value, _NotSetType): 361 self._completeIfNeeded() 362 363 def _completeIfNeeded(self) -> None: 364 if not self.__completed: 365 self.__complete() 366 367 def __complete(self) -> None: 368 if self._url.value is None: 369 raise IncompletableObject(400, message="Returned object contains no URL") 370 headers, data = self._requester.requestJsonAndCheck("GET", self._url.value) 371 self._storeAndUseAttributes(headers, data) 372 self.__completed = True 373 374 def update(self, additional_headers: Optional[Dict[str, Any]] = None) -> bool: 375 """ 376 Check and update the object with conditional request 377 :rtype: Boolean value indicating whether the object is changed 378 """ 379 conditionalRequestHeader = dict() 380 if self.etag is not None: 381 conditionalRequestHeader[Consts.REQ_IF_NONE_MATCH] = self.etag 382 if self.last_modified is not None: 383 conditionalRequestHeader[Consts.REQ_IF_MODIFIED_SINCE] = self.last_modified 384 if additional_headers is not None: 385 conditionalRequestHeader.update(additional_headers) 386 387 status, responseHeaders, output = self._requester.requestJson( 388 "GET", self._url.value, headers=conditionalRequestHeader 389 ) 390 if status == 304: 391 return False 392 else: 393 headers, data = self._requester._Requester__check(status, responseHeaders, output) # type: ignore 394 self._storeAndUseAttributes(headers, data) 395 self.__completed = True 396 return True