Initial release

Author: anthonyjb

README.md

@@ -30,7 +30,7 @@ suggestions = geouk.resources.Place.typeahead(client, 'wo...')
 # Fetch a list of `Place`s that match the given query string
 places = geouk.resources.Place.search(client, 'worcester')
 
-print(places.humanized_name, places[0].geo_coords)
+print(places[0].humanized_name, places[0].geo_coords)
 
 >> Worcester [52.18935, -2.22001]
 

geouk/__init__.py

@@ -0,0 +1,4 @@
+from .client import *
+
+from . import exceptions
+from . import resources

geouk/client.py

@@ -0,0 +1,127 @@
+import io
+import json
+
+import requests
+
+from . import exceptions
+
+__all__ = ['Client']
+
+
+class Client:
+    """
+    A client for the GeoUK API.
+    """
+
+    def __init__(self, api_key, api_base_url='https://api.geouk.xyz'):
+
+        # A key used to authenticate API calls to an account
+        self._api_key = api_key
+
+        # The base URL to use when calling the API
+        self._api_base_url = api_base_url
+
+        # NOTE: Rate limiting information is only available after a request
+        # has been made.
+
+        # The maximum number of requests per second that can be made with the
+        # given API key.
+        self._rate_limit = None
+
+        # The time (seconds since epoch) when the current rate limit will
+        # reset.
+        self._rate_limit_reset = None
+
+        # The number of requests remaining within the current limit before the
+        # next reset.
+        self._rate_limit_remaining = None
+
+    @property
+    def rate_limit(self):
+        return self._rate_limit
+
+    @property
+    def rate_limit_reset(self):
+        return self._rate_limit_reset
+
+    @property
+    def rate_limit_remaining(self):
+        return self._rate_limit_remaining
+
+    def __call__(self,
+        method,
+        path,
+        params=None,
+        data=None,
+        json_type_body=None,
+        files=None,
+        download=False
+    ):
+        """Call the API"""
+
+        # Build headers
+        headers = {'X-GeoUK-APIKey': self._api_key}
+
+        if json_type_body:
+            headers['Content-Type'] = 'application/json'
+
+        if not download:
+            headers['Accept'] = 'application/json'
+
+        if params:
+            # Filter out parameters set to `None`
+            params = {k: v for k, v in params.items() if v is not None}
+
+        # Make the request
+        r = getattr(requests, method.lower())(
+            f'{self._api_base_url}/{path}',
+            headers=headers,
+            params=params,
+            data=data,
+            json=json_type_body,
+            files=files
+        )
+
+        # Update the rate limit
+        if 'X-GeoUK-RateLimit-Limit' in r.headers:
+            self._rate_limit = int(r.headers['X-GeoUK-RateLimit-Limit'])
+            self._rate_limit_reset \
+                    = float(r.headers['X-GeoUK-RateLimit-Reset'])
+            self._rate_limit_remaining \
+                    = int(r.headers['X-GeoUK-RateLimit-Remaining'])
+
+        # Handle a successful response
+        if r.status_code in [200, 204]:
+
+            if download:
+                return io.BytesIO(r.content)
+
+            if r.headers.get('Content-Type', '')\
+                    .startswith('application/json'):
+
+                return r.json()
+
+            return None
+
+        # Raise an error related to the response
+        try:
+            error = r.json()
+
+        except json.decoder.JSONDecodeError:
+            pass
+
+        finally:
+            if not isinstance(error, dict):
+                error = {}
+
+        error_cls = exceptions.GeoUKException.get_class_by_status_code(
+            r.status_code
+        )
+
+        raise error_cls(
+            r.status_code,
+            error.get('hint'),
+            error.get('arg_errors')
+        )
+
+

geouk/exceptions.py

@@ -0,0 +1,109 @@
+import copy
+import inspect
+
+__all__ = [
+    'GeoUKException',
+    'GeoUKForbidden',
+    'GeoUKInvalidRequest',
+    'GeoUKNotFound',
+    'GeoUKRequestLimitExceeded',
+    'GeoUKUnauthorized'
+]
+
+
+class GeoUKException(Exception):
+    """
+    An error occurred while processing the request.
+    """
+
+    def __init__(self, status_code, hint=None, arg_errors=None):
+        super().__init__()
+
+        # The status code associated with the error
+        self.status_code = status_code
+
+        # A hint providing additional information as to why this error
+        # occurred.
+        self._hint = hint
+
+        # A dictionary of errors relating to the arguments (parameters) sent
+        # to the API endpoint (e.g `{'arg_name': ['error1', ...]}`).
+        self._arg_errors = arg_errors
+
+    def __str__(self):
+
+        doc_str = inspect.cleandoc(self.__class__.__doc__)
+        parts = [f'[{self.status_code}] {doc_str}']
+
+        if self._hint:
+            parts.append(f'Hint: {self._hint}')
+
+        if self._arg_errors:
+            parts.append(
+                'Argument errors:\n- ' + '\n- '.join([
+                    f'{arg_name}: {" ".join(errors)}'
+                    for arg_name, errors in self._arg_errors.items()
+                ])
+            )
+
+        return '\n---\n'.join(parts)
+
+    @property
+    def arg_errors(self):
+        if self._arg_errors:
+            return copy.deepcopy(self._arg_errors)
+
+    @property
+    def hint(self):
+        return self._hint
+
+    @classmethod
+    def get_class_by_status_code(cls, error_type, default=None):
+        """
+        Return the exception class associated with the status code, if no
+        class matches the given status code then the base `GeoUKException`
+        class is returned.
+        """
+
+        class_map = {
+            400: GeoUKInvalidRequest,
+            401: GeoUKUnauthorized,
+            403: GeoUKForbidden,
+            405: GeoUKForbidden,
+            404: GeoUKNotFound,
+            429: GeoUKRequestLimitExceeded
+        }
+
+        return class_map.get(error_type, default or GeoUKException)
+
+
+class GeoUKForbidden(GeoUKException):
+    """
+    The request is not not allowed, most likely the HTTP method used to call
+    the API endpoint is incorrect or the API key (via its associated account)
+    does not have permission to call the endpoint and/or perform the action.
+    """
+
+
+class GeoUKInvalidRequest(GeoUKException):
+    """
+    Not a valid request, most likely a missing or invalid parameter.
+    """
+
+
+class GeoUKNotFound(GeoUKException):
+    """
+    The endpoint you are calling or the document you referenced doesn't exist.
+    """
+
+
+class GeoUKRequestLimitExceeded(GeoUKException):
+    """
+    You have exceeded the number of API requests allowed per second.
+    """
+
+
+class GeoUKUnauthorized(GeoUKException):
+    """
+    The API key provided is not valid.
+    """

geouk/resources.py

@@ -0,0 +1,97 @@
+
+# NOTE: The `Place` class provides a thin wrappers to data fetched from the
+# API by the API client and should not be initialized directly.
+
+
+class _BaseResource:
+    """
+    A base resource used to wrap documents fetched from the API with dot
+    notation access to attributes and methods for access to related API
+    endpoints.
+    """
+
+    def __init__(self, client, document):
+
+        # The API client used to fetch the resource
+        self._client = client
+
+        # The document representing the resource's data
+        self._document = document
+
+    def __getattr__(self, name):
+
+        if '_document' in self.__dict__:
+            return self.__dict__['_document'].get(name, None)
+
+        raise AttributeError(
+            f"'{self.__class__.__name__}' has no attribute '{name}'"
+        )
+
+    def __getitem__(self, name):
+        return self.__dict__['_document'][name]
+
+    def __contains__(self, name):
+        return name in self.__dict__['_document']
+
+    def get(self, name, default=None):
+        return self.__dict__['_document'].get(name, default)
+
+
+
+class Place(_BaseResource):
+    """
+    A place within the UK or Republic of Ireland.
+    """
+
+    def __str__(self):
+        return f'Place: {self.humanized_name}'
+
+    @classmethod
+    def search(cls, client, q):
+        """
+        Fetch a list of places that match the given query string.
+        """
+
+        # Fetch the matching places
+        r = client(
+            'get',
+            'places/search',
+            params={'q': q}
+        )
+
+        return [cls(client, p) for p in r]
+
+    @classmethod
+    def typeahead(cls, client, q):
+        """
+        Return a list of place names and postcodes starting with the query
+        string's first 2 characters
+        """
+
+        # Fetch the typeahead results
+        r = client(
+            'get',
+            'places/typeahead',
+            params={'q': q}
+        )
+
+        return r
+
+class Source(_BaseResource):
+    """
+    A source for the data store against a place which provides attribution for
+    the data.
+    """
+
+    def __str__(self):
+        return f'Source: {self.name}'
+
+    @classmethod
+    def all(cls, client):
+        """Fetch a list of all sources"""
+        return [cls(client, s) for s in client('get', 'sources')]
+
+    @classmethod
+    def one(cls, client, ref):
+        """Return a source matching the given reference"""
+        return cls(client, client('get', f'sources/{ref}'))

setup.py

@@ -1,5 +1,5 @@
 """
-Setup instuctions for h51.
+Setup instuctions for GeoUK.
 """
 
 # Always prefer setuptools over distutils