calefaction/calefaction/eve/esi.py

# -*- coding: utf-8  -*-

from datetime import datetime
import random
from threading import Lock

import requests

from ..exceptions import EVEAPIError

__all__ = ["EVESwaggerInterface"]

class _ESICache:
    """Caches ESI API responses according to their headers.

    This interface is thread-safe.
    """
    EXPIRATION_RATE = 0.2

    def __init__(self):
        self._index = {}
        self._lock = Lock()

    @staticmethod
    def _calculate_expiry(resp):
        """Calculate the expiration date for the given response object."""
        if "Cache-Control" not in resp.headers:
            return None

        directives = resp.headers["Cache-Control"].lower().split(";")
        directives = [d.strip() for d in directives]
        if "public" not in directives:
            return None

        expires = resp.headers.get("Expires")
        if not expires:
            return None

        try:
            return datetime.strptime(expires, "%a, %d %b %Y %H:%M:%S GMT")
        except ValueError:
            return None

    @staticmethod
    def freeze_dict(d):
        """Return a hashable string key for the given dictionary."""
        if not d:
            return "{}"
        items = sorted((repr(str(k)), repr(str(v))) for k, v in d.items())
        return "{" + ",".join(":".join(p) for p in items) + "}"

    def _expire(self):
        """Remove old entries from the cache. Assumes lock is acquired."""
        condemned = []
        now = datetime.utcnow()
        for key, (expiry, _) in self._index.items():
            if expiry < now:
                condemned.append(key)
        for key in condemned:
            del self._index[key]

    def fetch(self, key):
        """Try to look up a key in the cache. Return None if not found.

        The key should be a string.

        This will periodically clear the cache of expired entries.
        """
        with self._lock:
            if random.random() < self.EXPIRATION_RATE:
                self._expire()
            if key in self._index:
                expiry, value = self._index[key]
                if expiry > datetime.utcnow():
                    return value
            return None

    def insert(self, key, value, response):
        """Store a key-value pair using the response as cache control."""
        expiry = self._calculate_expiry(response)
        if expiry and expiry > datetime.utcnow():
            with self._lock:
                self._index[key] = (expiry, value)


class _ESIQueryBuilder:
    """Stores an ESI query that is being built by the client."""

    def __init__(self, esi, token):
        self._esi = esi
        self._token = token
        self._path = "/"

    def __getattr__(self, item):
        self._path += str(item) + "/"
        return self

    def __call__(self, item):
        self._path += str(item) + "/"
        return self

    def get(self, **kwargs):
        """Do an HTTP GET request for the built query."""
        return self._esi.get(self._path, self._token, params=kwargs)

    def post(self, **kwargs):
        """Do an HTTP POST request for the built query."""
        return self._esi.post(self._path, self._token, data=kwargs)

    def put(self, **kwargs):
        """Do an HTTP PUT request for the built query."""
        return self._esi.put(self._path, self._token, data=kwargs)

    def delete(self, **kwargs):
        """Do an HTTP DELETE request for the built query."""
        return self._esi.delete(self._path, self._token, params=kwargs)


class EVESwaggerInterface:
    """EVE API module for the EVE Swagger Interface (ESI).

    There are two equivalent ways to use this interface:

    data = esi.get("/v3/characters/{char_id}/".format(char_id=char_id), token)
    data = esi(token).v3.characters(char_id).get()

    For more complex requests:

    data = esi.post("/v1/universe/names/", token, data={"ids": [entity_id]})
    data = esi(token).v1.universe.names.post(ids=[entity_id]})
    """

    def __init__(self, session):
        self._session = session
        self._base_url = "https://esi.tech.ccp.is"
        self._data_source = "tranquility"
        self._cache = _ESICache()

    def __call__(self, token):
        return _ESIQueryBuilder(self, token)

    def _do(self, method, query, params, data, token, can_cache=False):
        """Execute a query using a token with the given session method.

        Return the JSON result, if any. Raise EVEAPIError for any errors.
        """
        if can_cache:
            pkey = self._cache.freeze_dict(params)
            key = "|".join((method.__name__, self._data_source, query, pkey))
            cached = self._cache.fetch(key)
            if cached:
                return cached

        headers = {
            "Accept": "application/json",
            "Authorization": "Bearer " + token
        }
        params = params.copy() if params else {}
        params["datasource"] = self._data_source
        url = self._base_url + query

        try:
            resp = method(url, params=params, json=data or None,
                          headers=headers, timeout=10)
            resp.raise_for_status()
            result = resp.json() if resp.content else None
        except (requests.RequestException, ValueError) as exc:
            raise EVEAPIError(str(exc))

        if can_cache and result is not None:
            self._cache.insert(key, result, resp)
        return result

    def get(self, query, token, params=None):
        """Do an HTTP GET request for a query using a token."""
        meth = self._session.get
        return self._do(meth, query, params, None, token, True)

    def post(self, query, token, params=None, data=None):
        """Do an HTTP POST request for a query using a token."""
        meth = self._session.post
        return self._do(meth, query, params, data, token)

    def put(self, query, token, params=None, data=None):
        """Do an HTTP PUT request for a query using a token."""
        meth = self._session.put
        return self._do(meth, query, params, data, token)

    def delete(self, query, token, params=None):
        """Do an HTTP DELETE request for a query using a token."""
        meth = self._session.delete
        return self._do(meth, query, params, None, token)