Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Bower components Debian packages RPM packages NuGet packages

hemamaps / coreapi   python

Repository URL to install this package:

Version: 1.32.0 

/ client.py

from coreapi import codecs, exceptions, transports
from coreapi.compat import string_types
from coreapi.document import Document, Link
from coreapi.utils import determine_transport
import collections
import itypes


LinkAncestor = collections.namedtuple('LinkAncestor', ['document', 'keys'])


def _lookup_link(document, keys):
    """
    Validates that keys looking up a link are correct.

    Returns a two-tuple of (link, link_ancestors).
    """
    if not isinstance(keys, (list, tuple)):
        msg = "'keys' must be a list of strings or ints."
        raise TypeError(msg)
    if any([
        not isinstance(key, string_types) and not isinstance(key, int)
        for key in keys
    ]):
        raise TypeError("'keys' must be a list of strings or ints.")

    # Determine the link node being acted on, and its parent document.
    # 'node' is the link we're calling the action for.
    # 'document_keys' is the list of keys to the link's parent document.
    node = document
    link_ancestors = [LinkAncestor(document=document, keys=[])]
    for idx, key in enumerate(keys):
        try:
            node = node[key]
        except (KeyError, IndexError, TypeError):
            index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys)
            msg = 'Index %s did not reference a link. Key %s was not found.'
            raise exceptions.LinkLookupError(msg % (index_string, repr(key).strip('u')))
        if isinstance(node, Document):
            ancestor = LinkAncestor(document=node, keys=keys[:idx + 1])
            link_ancestors.append(ancestor)

    # Ensure that we've correctly indexed into a link.
    if not isinstance(node, Link):
        index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys)
        msg = "Can only call 'action' on a Link. Index %s returned type '%s'."
        raise exceptions.LinkLookupError(
            msg % (index_string, type(node).__name__)
        )

    return (node, link_ancestors)


class Client(itypes.Object):
    DEFAULT_TRANSPORTS = [
        transports.HTTPTransport()
    ]
    DEFAULT_DECODERS = [
        codecs.CoreJSONCodec(), codecs.JSONCodec(), codecs.TextCodec()
    ]

    def __init__(self, decoders=None, transports=None):
        if decoders is None:
            decoders = self.DEFAULT_DECODERS
        if transports is None:
            transports = self.DEFAULT_TRANSPORTS
        self._decoders = itypes.List(decoders)
        self._transports = itypes.List(transports)

    @property
    def decoders(self):
        return self._decoders

    @property
    def transports(self):
        return self._transports

    def get(self, url, force_codec=False):
        link = Link(url, action='get')

        # Perform the action, and return a new document.
        transport = determine_transport(self.transports, link.url)
        return transport.transition(link, self.decoders, force_codec=force_codec)

    def reload(self, document, force_codec=False):
        url = document.url
        link = Link(url, action='get')

        # Perform the action, and return a new document.
        transport = determine_transport(self.transports, link.url)
        return transport.transition(link, self.decoders, force_codec=force_codec)

    def action(self, document, keys, params=None, action=None, encoding=None, transform=None):
        if isinstance(keys, string_types):
            keys = [keys]

        # Validate the keys and link parameters.
        link, link_ancestors = _lookup_link(document, keys)

        if (action is not None) or (encoding is not None) or (transform is not None):
            # Handle any explicit overrides.
            action = link.action if (action is None) else action
            encoding = link.encoding if (encoding is None) else encoding
            transform = link.transform if (transform is None) else transform
            link = Link(link.url, action=action, encoding=encoding, transform=transform, fields=link.fields)

        # Perform the action, and return a new document.
        transport = determine_transport(self.transports, link.url)
        return transport.transition(link, self.decoders, params=params, link_ancestors=link_ancestors)