mirror of
https://github.com/status-im/PyGithub.git
synced 2026-08-31 19:01:15 +00:00
If a returned object from GitHub does not include an URL, that object can not be completed, and used to fail with an obscure traceback saying NoneType has no method startswith. Check for this, and raise IncompletableObject instead.
301 lines
11 KiB
Python
301 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
############################ Copyrights and license ############################
|
|
# #
|
|
# Copyright 2012 Vincent Jacques <vincent@vincent-jacques.net> #
|
|
# Copyright 2012 Zearin <zearin@gonk.net> #
|
|
# Copyright 2013 AKFish <akfish@gmail.com> #
|
|
# Copyright 2013 Vincent Jacques <vincent@vincent-jacques.net> #
|
|
# Copyright 2014 Andrew Scheller <github@loowis.durge.org> #
|
|
# Copyright 2014 Vincent Jacques <vincent@vincent-jacques.net> #
|
|
# Copyright 2016 Jakub Wilk <jwilk@jwilk.net> #
|
|
# Copyright 2016 Jannis Gebauer <ja.geb@me.com> #
|
|
# Copyright 2016 Peter Buckley <dx-pbuckley@users.noreply.github.com> #
|
|
# Copyright 2016 Sam Corbett <sam.corbett@cloudsoftcorp.com> #
|
|
# Copyright 2018 sfdye <tsfdye@gmail.com> #
|
|
# #
|
|
# This file is part of PyGithub. #
|
|
# http://pygithub.readthedocs.io/ #
|
|
# #
|
|
# PyGithub is free software: you can redistribute it and/or modify it under #
|
|
# the terms of the GNU Lesser General Public License as published by the Free #
|
|
# Software Foundation, either version 3 of the License, or (at your option) #
|
|
# any later version. #
|
|
# #
|
|
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
|
|
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
|
|
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
|
|
# details. #
|
|
# #
|
|
# You should have received a copy of the GNU Lesser General Public License #
|
|
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
|
|
# #
|
|
################################################################################
|
|
|
|
from __future__ import absolute_import
|
|
import datetime
|
|
from operator import itemgetter
|
|
|
|
from . import GithubException
|
|
from . import Consts
|
|
import six
|
|
|
|
|
|
class _NotSetType:
|
|
def __repr__(self):
|
|
return "NotSet"
|
|
|
|
value = None
|
|
|
|
|
|
NotSet = _NotSetType()
|
|
|
|
|
|
class _ValuedAttribute:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
|
|
class _BadAttribute:
|
|
def __init__(self, value, expectedType, exception=None):
|
|
self.__value = value
|
|
self.__expectedType = expectedType
|
|
self.__exception = exception
|
|
|
|
@property
|
|
def value(self):
|
|
raise GithubException.BadAttributeException(self.__value, self.__expectedType, self.__exception)
|
|
|
|
|
|
class GithubObject(object):
|
|
"""
|
|
Base class for all classes representing objects returned by the API.
|
|
"""
|
|
|
|
'''
|
|
A global debug flag to enable header validation by requester for all objects
|
|
'''
|
|
CHECK_AFTER_INIT_FLAG = False
|
|
|
|
@classmethod
|
|
def setCheckAfterInitFlag(cls, flag):
|
|
cls.CHECK_AFTER_INIT_FLAG = flag
|
|
|
|
def __init__(self, requester, headers, attributes, completed):
|
|
self._requester = requester
|
|
self._initAttributes()
|
|
self._storeAndUseAttributes(headers, attributes)
|
|
|
|
# Ask requester to do some checking, for debug and test purpose
|
|
# Since it's most handy to access and kinda all-knowing
|
|
if self.CHECK_AFTER_INIT_FLAG: # pragma no branch (Flag always set in tests)
|
|
requester.check_me(self)
|
|
|
|
def _storeAndUseAttributes(self, headers, attributes):
|
|
# Make sure headers are assigned before calling _useAttributes
|
|
# (Some derived classes will use headers in _useAttributes)
|
|
self._headers = headers
|
|
self._rawData = attributes
|
|
self._useAttributes(attributes)
|
|
|
|
@property
|
|
def raw_data(self):
|
|
"""
|
|
:type: dict
|
|
"""
|
|
self._completeIfNeeded()
|
|
return self._rawData
|
|
|
|
@property
|
|
def raw_headers(self):
|
|
"""
|
|
:type: dict
|
|
"""
|
|
self._completeIfNeeded()
|
|
return self._headers
|
|
|
|
@staticmethod
|
|
def _parentUrl(url):
|
|
return "/".join(url.split("/")[: -1])
|
|
|
|
@staticmethod
|
|
def __makeSimpleAttribute(value, type):
|
|
if value is None or isinstance(value, type):
|
|
return _ValuedAttribute(value)
|
|
else:
|
|
return _BadAttribute(value, type)
|
|
|
|
@staticmethod
|
|
def __makeSimpleListAttribute(value, type):
|
|
if isinstance(value, list) and all(isinstance(element, type) for element in value):
|
|
return _ValuedAttribute(value)
|
|
else:
|
|
return _BadAttribute(value, [type])
|
|
|
|
@staticmethod
|
|
def __makeTransformedAttribute(value, type, transform):
|
|
if value is None:
|
|
return _ValuedAttribute(None)
|
|
elif isinstance(value, type):
|
|
try:
|
|
return _ValuedAttribute(transform(value))
|
|
except Exception as e:
|
|
return _BadAttribute(value, type, e)
|
|
else:
|
|
return _BadAttribute(value, type)
|
|
|
|
@staticmethod
|
|
def _makeStringAttribute(value):
|
|
return GithubObject.__makeSimpleAttribute(value, (str, six.text_type))
|
|
|
|
@staticmethod
|
|
def _makeIntAttribute(value):
|
|
return GithubObject.__makeSimpleAttribute(value, six.integer_types)
|
|
|
|
@staticmethod
|
|
def _makeBoolAttribute(value):
|
|
return GithubObject.__makeSimpleAttribute(value, bool)
|
|
|
|
@staticmethod
|
|
def _makeDictAttribute(value):
|
|
return GithubObject.__makeSimpleAttribute(value, dict)
|
|
|
|
@staticmethod
|
|
def _makeTimestampAttribute(value):
|
|
return GithubObject.__makeTransformedAttribute(value, six.integer_types, datetime.datetime.utcfromtimestamp)
|
|
|
|
@staticmethod
|
|
def _makeDatetimeAttribute(value):
|
|
def parseDatetime(s):
|
|
if len(s) == 24: # pragma no branch (This branch was used only when creating a download)
|
|
# The Downloads API has been removed. I'm keeping this branch because I have no mean
|
|
# to check if it's really useless now.
|
|
return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%S.000Z") # pragma no cover (This branch was used only when creating a download)
|
|
elif len(s) >= 25:
|
|
return datetime.datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S") + (1 if s[19] == '-' else -1) * datetime.timedelta(hours=int(s[20:22]), minutes=int(s[23:25]))
|
|
else:
|
|
return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
return GithubObject.__makeTransformedAttribute(value, (str, six.text_type), parseDatetime)
|
|
|
|
def _makeClassAttribute(self, klass, value):
|
|
return GithubObject.__makeTransformedAttribute(value, dict, lambda value: klass(self._requester, self._headers, value, completed=False))
|
|
|
|
@staticmethod
|
|
def _makeListOfStringsAttribute(value):
|
|
return GithubObject.__makeSimpleListAttribute(value, (str, six.text_type))
|
|
|
|
@staticmethod
|
|
def _makeListOfIntsAttribute(value):
|
|
return GithubObject.__makeSimpleListAttribute(value, int)
|
|
|
|
@staticmethod
|
|
def _makeListOfDictsAttribute(value):
|
|
return GithubObject.__makeSimpleListAttribute(value, dict)
|
|
|
|
@staticmethod
|
|
def _makeListOfListOfStringsAttribute(value):
|
|
return GithubObject.__makeSimpleListAttribute(value, list)
|
|
|
|
def _makeListOfClassesAttribute(self, klass, value):
|
|
if isinstance(value, list) and all(isinstance(element, dict) for element in value):
|
|
return _ValuedAttribute([klass(self._requester, self._headers, element, completed=False) for element in value])
|
|
else:
|
|
return _BadAttribute(value, [dict])
|
|
|
|
def _makeDictOfStringsToClassesAttribute(self, klass, value):
|
|
if isinstance(value, dict) and all(isinstance(key, (str, six.text_type)) and isinstance(element, dict) for key, element in six.iteritems(value)):
|
|
return _ValuedAttribute(dict((key, klass(self._requester, self._headers, element, completed=False)) for key, element in six.iteritems(value)))
|
|
else:
|
|
return _BadAttribute(value, {(str, six.text_type): dict})
|
|
|
|
@property
|
|
def etag(self):
|
|
'''
|
|
:type: str
|
|
'''
|
|
return self._headers.get(Consts.RES_ETAG)
|
|
|
|
@property
|
|
def last_modified(self):
|
|
'''
|
|
:type: str
|
|
'''
|
|
return self._headers.get(Consts.RES_LAST_MODIFIED)
|
|
|
|
def get__repr__(self, params):
|
|
"""
|
|
Converts the object to a nicely printable string.
|
|
"""
|
|
def format_params(params):
|
|
items = list(params.items())
|
|
for k, v in sorted(items, key=itemgetter(0), reverse=True):
|
|
if isinstance(v, bytes):
|
|
v = v.decode('utf-8')
|
|
if isinstance(v, six.text_type):
|
|
v = u'"{v}"'.format(v=v)
|
|
yield u'{k}={v}'.format(k=k, v=v)
|
|
return u'{class_name}({params})'.format(
|
|
class_name=self.__class__.__name__,
|
|
params=u", ".join(list(format_params(params)))
|
|
)
|
|
|
|
|
|
class NonCompletableGithubObject(GithubObject):
|
|
def _completeIfNeeded(self):
|
|
pass
|
|
|
|
|
|
class CompletableGithubObject(GithubObject):
|
|
def __init__(self, requester, headers, attributes, completed):
|
|
GithubObject.__init__(self, requester, headers, attributes, completed)
|
|
self.__completed = completed
|
|
|
|
def __eq__(self, other):
|
|
return other.__class__ is self.__class__ and other._url.value == self._url.value
|
|
|
|
def __ne__(self, other):
|
|
return not self == other
|
|
|
|
def _completeIfNotSet(self, value):
|
|
if value is NotSet:
|
|
self._completeIfNeeded()
|
|
|
|
def _completeIfNeeded(self):
|
|
if not self.__completed:
|
|
self.__complete()
|
|
|
|
def __complete(self):
|
|
if self._url.value is None:
|
|
raise GithubException.IncompletableObject(400, "Returned object contains no URL")
|
|
headers, data = self._requester.requestJsonAndCheck(
|
|
"GET",
|
|
self._url.value
|
|
)
|
|
self._storeAndUseAttributes(headers, data)
|
|
self.__completed = True
|
|
|
|
def update(self):
|
|
'''
|
|
Check and update the object with conditional request
|
|
:rtype: Boolean value indicating whether the object is changed
|
|
'''
|
|
conditionalRequestHeader = dict()
|
|
if self.etag is not None:
|
|
conditionalRequestHeader[Consts.REQ_IF_NONE_MATCH] = self.etag
|
|
if self.last_modified is not None:
|
|
conditionalRequestHeader[Consts.REQ_IF_MODIFIED_SINCE] = self.last_modified
|
|
|
|
status, responseHeaders, output = self._requester.requestJson(
|
|
"GET",
|
|
self._url.value,
|
|
headers=conditionalRequestHeader
|
|
)
|
|
if status == 304:
|
|
return False
|
|
else:
|
|
headers, data = self._requester._Requester__check(status, responseHeaders, output)
|
|
self._storeAndUseAttributes(headers, data)
|
|
self.__completed = True
|
|
return True
|