# -*- coding: utf-8 -*- ############################ Copyrights and license ############################ # # # Copyright 2012 Vincent Jacques # # Copyright 2012 Zearin # # Copyright 2013 AKFish # # Copyright 2013 Vincent Jacques # # # # This file is part of PyGithub. http://jacquev6.github.com/PyGithub/ # # # # 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 . # # # ################################################################################ import datetime import GithubException import Consts class _NotSetType: def __repr__(self): return "NotSet" NotSet = _NotSetType() 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 _NoneIfNotSet(value): if value is NotSet: return None else: return value @staticmethod def _parseDatetime(s): if s is None: return None elif len(s) == 24: return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%S.000Z") 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") @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_MODIFED) 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 _completeIfNotSet(self, value): if value is NotSet: self._completeIfNeeded() def _completeIfNeeded(self): if not self.__completed: self.__complete() def __complete(self): headers, data = self._requester.requestJsonAndCheck( "GET", self._url ) 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, 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