Project import

.gitignore

@@ -0,0 +1,4 @@
+*.pyc
+githubcap.egg-info
+build
+dist

Pipfile

@@ -0,0 +1,19 @@
+[[source]]
+
+url = "https://pypi.python.org/simple"
+verify_ssl = true
+name = "pypi"
+
+
+[packages]
+
+attrs = "*"
+requests = "*"
+click = "*"
+daiquiri = "*"
+voluptuous = "*"
+pyyaml = "*"
+
+
+[dev-packages]
+

githubcap-cli

@@ -0,0 +1 @@
+githubcap/cli.py

githubcap/__init__.py

@@ -0,0 +1,12 @@
+from .configuration import Configuration
+from .enums import Filtering
+from .enums import State
+from .enums import Sorting
+from .enums import SortingDirection
+from .utils import setup_logging
+
+__version__ = '1.0.0rc1'
+__title__ = 'githubcap'
+__author__ = 'Fridolin Pokorny'
+__license__ = 'ASL 2.0'
+__copyright__ = 'Copyright 2018 Fridolin Pokorny'

githubcap/base.py

@@ -0,0 +1,181 @@
+import attr
+from datetime import datetime
+import copy
+import enum
+import logging
+import requests
+import time
+import typing
+
+from voluptuous import Schema
+
+from .utils import parse_datetime
+from .utils import serialize_datetime
+from .configuration import Configuration
+from .exceptions import MissingPassword
+from .exceptions import HTTPError
+from .utils import next_pagination_page
+from .enums import GitHubCapEnum
+
+_LOG = logging.getLogger(__name__)
+
+
+@attr.s
+class GitHubHandlerBase(object):
+    DEFAULT_PER_PAGE: typing.ClassVar[int] = Configuration().per_page_listing
+    config: typing.ClassVar[Configuration] = Configuration()
+
+    @classmethod
+    def call(cls, uri, payload=None, method=None):
+        requests_kwargs = {
+            'headers': copy.copy(Configuration().headers)
+        }
+
+        if Configuration().token:
+            _LOG.debug("Using OAuth2 token '%s***' for GitHub call", Configuration().token[:4])
+            requests_kwargs['headers']['Authorization'] = 'token {!s}'.format(Configuration().token)
+        elif Configuration().user:
+            if not Configuration().password:
+                raise MissingPassword("No password set for user {!s}".format(Configuration().user))
+
+            _LOG.debug("Using basic authentication for user %s", Configuration().user)
+            requests_kwargs['auth'].append((Configuration().user, Configuration().password))
+        else:
+            _LOG.debug("No authentication is used")
+
+        url = "{!s}/{!s}".format(Configuration().github_api, uri)
+        requests_func = getattr(requests, method.lower())
+        if payload:
+            requests_kwargs['json'] = payload
+
+        while True:
+            _LOG.debug("%s %s", method, url)
+            response = requests_func(url, **requests_kwargs)
+
+            _LOG.debug("Request took %s and the HTTP status code for response was %d",
+                       response.elapsed, response.status_code)
+
+            if not (response.status_code == 403 and
+                    response.json()['message'].startswith("API rate limit exceeded") and
+                    cls.config.omit_rate_limiting):
+                break
+
+            reset_datetime = datetime.fromtimestamp(int(response.headers['X-RateLimit-Reset']))
+            sleep_time = (reset_datetime - datetime.now()).total_seconds()
+            _LOG.debug("API rate limit hit, retrying in %d seconds...", sleep_time)
+            time.sleep(sleep_time)
+
+        try:
+            # Rely on request's checks here
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as exc:
+            raise HTTPError(response.json(), response.status_code) from exc
+
+        return response.json(), response.headers
+
+    @classmethod
+    def head(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='HEAD')
+
+    @classmethod
+    def get(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='GET')
+
+    @classmethod
+    def post(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='POST')
+
+    @classmethod
+    def patch(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='PATCH')
+
+    @classmethod
+    def put(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='PUT')
+
+    @classmethod
+    def delete(cls, *args, **kwargs):
+        return cls.call(*args, **kwargs, method='DELETE')
+
+    def _do_listing(self, base_uri):
+        while True:
+            uri = '{!s}?{!s}'.format(base_uri, self._get_query_string())
+            response, headers = self.get(uri)
+
+            for entry in response:
+                yield entry
+
+            if not self.config.pagination:
+                return
+
+            next_page = next_pagination_page(headers)
+            if next_page is None:
+                return
+            self.page = next_page
+
+    @classmethod
+    def submit(cls, item):
+        raise NotImplementedError
+
+
+class GitHubBase(object):
+    _SCHEMA: typing.ClassVar[Schema] = None
+
+    # TODO: dirty flag
+
+    @classmethod
+    def from_response(cls, response):
+        if cls._SCHEMA is None:
+            raise NotImplementedError("No schema defined for entity {!r}".format(cls.__name__))
+
+        if Configuration().validate_schemas:
+            cls._SCHEMA(response)
+
+        return cls.from_dict(response)
+
+    @classmethod
+    def _from_dict_value(cls, attribute_type, value):
+        if attribute_type == datetime:
+            return parse_datetime(value)
+        elif issubclass(attribute_type, GitHubCapEnum):
+            return attribute_type.from_value(value)
+        elif issubclass(attribute_type, GitHubBase):
+            return attribute_type.from_dict(value)
+        elif '_gorg' in attribute_type.__dict__ and attribute_type._gorg == typing.List:
+            assert len(attribute_type.__args__) == 1, "Type defined multiple types: {!r}".format(attribute_type)
+            return list(cls._from_dict_value(attribute_type.__args__[0], item) for item in value)
+
+        return value
+
+    @classmethod
+    def from_dict(cls, dict_):
+        if dict_ is None:
+            return None
+
+        values = {}
+        for attribute in cls.__attrs_attrs__:
+            if attribute.name in dict_:
+                values[attribute.name] = cls._from_dict_value(attribute.type, dict_[attribute.name])
+
+        return cls(**values)
+
+    @classmethod
+    def _to_dict_value(cls, value):
+        if isinstance(value, datetime):
+            return serialize_datetime(value)
+        elif isinstance(value, GitHubBase):
+            return value.to_dict()
+        elif issubclass(value.__class__, enum.Enum):
+            return value.value
+        elif isinstance(value, list):
+            return list(cls._to_dict_value(item) for item in value)
+
+        return value
+
+    def to_dict(self):
+        result = {}
+        for attribute in self.__attrs_attrs__:
+            result[attribute.name] = self._to_dict_value(getattr(self, attribute.name))
+        return result
+
+