Source code for civis.resources._resources

from collections import OrderedDict
from functools import lru_cache
import json
import os
import re
import textwrap
    from inspect import Signature, Parameter
except ImportError:
    from funcsigs import Signature, Parameter

from jsonref import JsonRef
import requests
from requests import Request

from civis.base import Endpoint, get_base_url
from civis._deprecation import deprecate_param
from civis._utils import (camel_to_snake, to_camelcase,
                          open_session, get_api_key,
                          retry_request, MAX_RETRIES)

API_VERSIONS = ["1.0"]

TYPE_MAP = {"array": "list", "object": "dict"}
    "iterator : bool, optional\n"
    "    If True, return a generator to iterate over all responses. Use when\n"
    "    more results than the maximum allowed by limit are needed. When\n"
    "    True, limit and page_num are ignored. Defaults to False.\n")
CACHED_SPEC_PATH = os.path.join(os.path.expanduser('~'),

@deprecate_param('v2.0.0', 'resources')
def exclude_resource(path, api_version, resources):
    if api_version == "1.0" and resources == "base":
        include = any([path.startswith(x) for x in BASE_RESOURCES_V1])
        include = True
    return not include

def get_properties(x):
    return x.get("properties") or x.get("items", {}).get("properties")

def property_type(props):
    t = TYPE_MAP.get(props['type'], props['type'])
    fmt = props.get("format")
    return "{}/{}".format(t, fmt) if fmt else t

def name_and_type_doc(name, prop, child, level, optional=False):
    """ Create a doc string element that includes a parameter's name
    and its type. This is intented to be combined with another
    doc string element that gives a description of the parameter.
    prop_type = property_type(prop)
    snake_name = camel_to_snake(name)
    indent = " " * level * 4
    dash = "- " if level > 0 else ""
    colons = "::" if child else ""
    opt_str = ", optional" if optional else ""
    doc = "{}{}{} : {}{}{}"
    return doc.format(indent, dash, snake_name, prop_type, opt_str, colons)

def docs_from_property(name, prop, properties, level, optional=False):
    """ Create a list of doc string elements from a single property
    object. Avoids infinite recursion when a property contains a
    circular reference to its parent.
    docs = []
    child_properties = get_properties(prop)
    child = None if child_properties == properties else child_properties
    docs.append(name_and_type_doc(name, prop, child, level, optional))
    doc_str = prop.get("description")
    if doc_str:
        indent = 4 * (level + 1) * " "
        doc_wrap = textwrap.fill(doc_str,
    if child:
        child_docs = docs_from_properties(child, level + 1)
    return docs

def docs_from_properties(properties, level=0):
    """ Return doc string elements from a dictionary of properties."""
    docs = []
    for name, prop in properties.items():
        doc_list = docs_from_property(name, prop, properties, level)
    return docs

def deprecated_notice(deprecation_warning):
    """ Return a doc string element for the deprecation notice. The
    doc string can be an empty string if the warning is None
    if deprecation_warning is None:
        return ""

    return "Deprecation warning!\n------------------\n" + deprecation_warning

def doc_from_responses(responses):
    """ Return a doc string element from a responses object. The
    doc string describes the returned objects of a function.
    response_code, response_object = next(iter(responses.items()))
    schema = response_object.get('schema', {})
    properties = get_properties(schema)
    if properties:
        result_doc = "\n".join(docs_from_properties(properties))
        description = response_object['description']
        result_doc_fmt = "None\n    Response code {}: {}"
        result_doc = result_doc_fmt.format(response_code, description)
    return "Returns\n-------\n" + result_doc

def join_doc_elements(*args):
    return "\n".join(args).rstrip()

def doc_from_param(param):
    """ Return a doc string element for a single parameter.
    Intended to be joined with other doc string elements to
    form a complete docstring of the accepted parameters of
    a function.
    snake_name = camel_to_snake(param['name'])
    param_type = param['type']
    desc = param.get('description')
    optional = "" if param["required"] else ", optional"
    doc_body = ""
    if desc:
        indent = " " * 4
        doc_wrap = textwrap.fill(desc,
        doc_body += doc_wrap
        doc_body += "\n"
    doc_head = "{} : {}{}\n".format(snake_name, param_type, optional)
    return doc_head + doc_body

def iterable_method(method, params):
    """Determine whether it is possible for this endpoint to return an iterated
    required_params = ('limit', 'page_num')
    params_present = all(param in params for param in required_params)
    return (method.lower() == 'get' and params_present)

def create_signature(args, optional_args):
    """ Dynamically create a signature for a function from strings.

    This function can be used to create a signature for a dynamically
    generated function without generating a string representation of
    the function code and making an explicit eval call.

    args : list
        List of strings that name the required arguments of a function.
    optional_args : list
        List of strings that name the optional arguments of a function.

    Signature(p) : inspect.Signature instance
        A Signature object that can be used to validate arguments in
        a dynamically created function.
    p = [Parameter(x, Parameter.POSITIONAL_OR_KEYWORD) for x in args]
    p += [Parameter(x, Parameter.KEYWORD_ONLY, default='DEFAULT')
          for x in optional_args]
    return Signature(p)

def split_method_params(params):
    args = []
    optional_args = []
    body_params = []
    query_params = []
    path_params = []
    for param in params:
        name = param["name"]
        if param["required"]:
        if param["in"] == "body":
        elif param["in"] == "query":
        elif param["in"] == "path":
    return args, optional_args, body_params, query_params, path_params

def create_method(params, verb, method_name, path, doc):
    """ Dynamically create a function to make an API call.

    The returned function accepts required parameters as positional arguments
    and optional parameters as kwargs.  The function passes these parameters
    into the appropriate place (path, query or body) in the API call,
    depending on the criteria of the API endpoint.

    params : list
        A list of dictionaries that specify the arguments used
        by the returned function f
    verb : str
        HTTP verb to call
    method_name : str
        The name to give the returned function f
    path : str
        Endpoint path, possibly including replacement fields
        (i.e. scripts/{id})
    doc : str
        Documentation string for the returned function f

    f : function
        A function which will make an API call
    elements = split_method_params(params)
    sig_args, sig_opt_args, body_params, query_params, path_params = elements
    sig = create_signature(sig_args, sig_opt_args)
    is_iterable = iterable_method(verb, query_params)

    def f(self, *args, **kwargs):
        raise_for_unexpected_kwargs(method_name, kwargs, sig_args,
                                    sig_opt_args, is_iterable)

        iterator = kwargs.pop('iterator', False)
        arguments = sig.bind(*args, **kwargs).arguments
        if arguments.get("kwargs"):
        body = {x: arguments[x] for x in body_params if x in arguments}
        query = {x: arguments[x] for x in query_params if x in arguments}
        path_vals = {x: arguments[x] for x in path_params if x in arguments}
        url = path.format(**path_vals) if path_vals else path
        return self._call_api(verb, url, query, body, iterator=iterator)

    # Add signature to function, including 'self' for class method
    sig_self = create_signature(["self"] + sig_args, sig_opt_args)
    f.__signature__ = sig_self
    f.__doc__ = doc
    f.__name__ = str(method_name)
    return f

def raise_for_unexpected_kwargs(method_name, arguments, sig_args, sig_kwargs,
    """Raise TypeError if arguments are not in sig_args or sig_kwargs."""
    expected = set(sig_args) | set(sig_kwargs)
    if is_iterable:
        expected |= {'iterator'}
    unexpected = set(arguments.keys()) - expected
    if unexpected:
        msg_fmt = "{}() got an unexpected keyword argument(s) {}"
        raise TypeError(msg_fmt.format(method_name, str(unexpected)))

def bracketed(x):
    return"^{.*}$", x)

def parse_param(param):
    """ Parse a parameter into a list of dictionaries which can
    be used to add the parameter to a dynamically generated function.
    doc = ""
    args = []
    param_in = param['in']
    if param_in == 'body':
        body_args = parse_param_body(param)
        snake_name = camel_to_snake(param['name'])
        req = param['required']
        doc = doc_from_param(param)
        a = {"name": snake_name, "in": param_in, "required": req, "doc": doc}
    return args

def parse_params(parameters, summary, verb):
    """ Parse the parameters of a function specification into a list
    of dictionaries which are used to generate the function at runtime.
    args = []
    for param in parameters:
    if iterable_method(verb, (x["name"] for x in args)):
        iter_arg = {"name": "iterator", "in": None,
                    "required": False, "doc": ITERATOR_PARAM_DESC}
    req_docs = [x["doc"] for x in args if x["required"]]
    opt_docs = [x["doc"] for x in args if not x["required"]]
    param_docs = "".join(req_docs + opt_docs)
    if summary:
        summary_str = "{}\n".format(textwrap.fill(summary, width=79))
        summary_str = ""
    if param_docs:
        docs = "{}\nParameters\n----------\n{}".format(summary_str, param_docs)
    elif summary:
        docs = summary_str
    return args, docs

def parse_param_body(parameter):
    """ Parse the nested element of a parameter into a list of dictionaries
    which can be used to add the parameter to a dynamically generated
    schema = parameter['schema']
    properties = schema['properties']
    req = schema.get('required', [])
    arguments = []
    for name, prop in properties.items():
        snake_name = camel_to_snake(name)
        is_req = name in req
        doc_list = docs_from_property(name, prop, properties, 0, not is_req)
        doc = "\n".join(doc_list) + "\n"
        a = {"name": snake_name, "in": "body", "required": is_req, "doc": doc}
    return arguments

def parse_method_name(verb, path):
    """ Create method name from endpoint path

    Create method name as the http verb (method) followed by
    any static path parameters. As there is ambiguity where
    a path may optionally have a path parameter passed (i.e.
    GET reports/ and GET reports/{id}), any GET method is changed
    to "list" for endpoints where the final path parameter is

    >>> parse_method_name("get", "")
    >>> parse_method_name("get", "{id}")
    >>> parse_method_name("get", "{id}/runs/{run_id}")
    >>> parse_method_name("post", "containers/{id}/runs/{run_id}")
    path_elems = path.split("/")[1:]
    name_elems = []
    for i, elem in enumerate(path_elems):
        prev_elem = path_elems[i - 1] if i > 0 else None
        if not bracketed(elem):
        elif prev_elem and bracketed(prev_elem):
    final_elem = path_elems[-1] if path_elems else ""
    verb = "list" if verb == "get" and (not bracketed(final_elem)) else verb
    path_name = "_".join(name_elems)
    method_name = "_".join((verb, path_name)) if path_name else verb
    return re.sub("-", "_", method_name)

def parse_method(verb, operation, path):
    """ Generate a python function from a specification of that function."""
    summary = operation["summary"]
    params = operation.get("parameters", [])
    responses = operation["responses"]
    deprecation_warning = operation.get("x-deprecation-warning", None)
    if 'deprecated' in summary.lower():
        return None

    args, param_doc = parse_params(params, summary, verb)
    response_doc = doc_from_responses(responses)
    deprecation_notice = deprecated_notice(deprecation_warning)
    docs = join_doc_elements(deprecation_notice, param_doc, response_doc)
    name = parse_method_name(verb, path)

    method = create_method(args, verb, name, path, docs)
    return name, method

@deprecate_param('v2.0.0', 'resources')
def parse_path(path, operations, api_version, resources):
    """ Parse an endpoint into a class where each valid http request
    on that endpoint is converted into a convenience function and
    attached to the class as a method.
    path = path.strip('/')
    modified_base_path = re.sub("-", "_", path.split('/')[0].lower())
    methods = []
    if exclude_resource(path, api_version, resources):
        return modified_base_path, methods
    for verb, op in operations.items():
        method = parse_method(verb, op, path)
        if method is None:
    return modified_base_path, methods

@deprecate_param('v2.0.0', 'resources')
def parse_api_spec(api_spec, api_version, resources):
    """ Dynamically create classes to interface with the Civis API.

    Parse an OpenAPI (Swagger) specification into a dictionary of classes
    where each class represents an endpoint resource and contains
    methods to make http requests on that resource.

    api_spec : OrderedDict
        The Civis API specification to parse.  References should be resolved
        before passing, typically using jsonref.JsonRef().
    api_version : string, optional
        The version of endpoints to call. May instantiate multiple client
        objects with different versions.  Currently only "1.0" is supported.
    resources : string, optional
        When set to "base", only the default endpoints will be exposed in the
        client object.  Set to "all" to include all endpoints available for
        a given user, including those that may be in development and subject
        to breaking changes at a later date.
    paths = api_spec['paths']
    classes = {}
    for path, ops in paths.items():
        base_path, methods = parse_path(path, ops, api_version, resources)
        class_name = to_camelcase(base_path)
        if methods and classes.get(base_path) is None:
            classes[base_path] = type(str(class_name), (Endpoint,), {})
        for method_name, method in methods:
            setattr(classes[base_path], method_name, method)
    return classes

def get_api_spec(api_key, api_version="1.0", user_agent="civis-python"):
    """Download the Civis API specification.

    api_key : str
        Your API key obtained from the Civis Platform.
    api_version : string, optional
        The version of endpoints to call. May instantiate multiple client
        objects with different versions.  Currently only "1.0" is supported.
    user_agent : string, optional
        Provide this user agent to the the Civis API, along with an
        API client version tag and ``requests`` version tag.
    if api_version == "1.0":
        with open_session(api_key, user_agent=user_agent) as sess:
            request = Request('GET', "{}endpoints".format(get_base_url()))
            pre_request = sess.prepare_request(request)
            response = retry_request('get', pre_request, sess, MAX_RETRIES)
        msg = "API specification for api version {} cannot be found"
        raise ValueError(msg.format(api_version))
    if response.status_code in (401, 403):
        msg = "{} error downloading API specification. API key may be expired."
        raise requests.exceptions.HTTPError(msg.format(response.status_code))
    spec = response.json(object_pairs_hook=OrderedDict)
    return spec

@deprecate_param('v2.0.0', 'resources')
def generate_classes(api_key, api_version="1.0", resources="all"):
    """ Dynamically create classes to interface with the Civis API.

    The Civis API documents behavior using an OpenAPI/Swagger specification.
    This function parses the specification and returns a dictionary of
    classes to provide a convenient method to make requests to the Civis
    API and to view documentation.

    api_key : str
        Your API key obtained from the Civis Platform.
    api_version : string, optional
        The version of endpoints to call. May instantiate multiple client
        objects with different versions.  Currently only "1.0" is supported.
    resources : string, optional
        When set to "base", only the default endpoints will be exposed in the
        client object.  Set to "all" to include all endpoints available for
        a given user, including those that may be in development and subject
        to breaking changes at a later date.
    assert api_version in API_VERSIONS, (
        "APIClient api_version must be one of {}".format(API_VERSIONS))
    assert resources in ["base", "all"], (
        "resources must be one of {}".format(["base", "all"]))
    raw_spec = get_api_spec(api_key, api_version)
    spec = JsonRef.replace_refs(raw_spec)
    return parse_api_spec(spec, api_version, resources)

def cache_api_spec(cache=CACHED_SPEC_PATH, api_key=None, api_version="1.0"):
    """Cache a local copy of the Civis Data Science API spec

    cache : str, optional
        File in which to store the cache of the API spec
    api_key : str, optional
        Your API key obtained from the Civis Platform. If not given, the
        client will use the :envvar:`CIVIS_API_KEY` environment variable.
    api_version : string, optional
        The version of endpoints to call. May instantiate multiple client
        objects with different versions.  Currently only "1.0" is supported.
    api_key = get_api_key(api_key)
    spec = get_api_spec(api_key, api_version=api_version)
    with open(cache, "wt") as _fout:
        json.dump(spec, _fout)

@deprecate_param('v2.0.0', 'resources')
def generate_classes_maybe_cached(cache, api_key, api_version, resources):
    """Generate class objects either from /endpoints or a local cache."""
    if cache is None:
        classes = generate_classes(api_key, api_version, resources)
        if isinstance(cache, OrderedDict):
            raw_spec = cache
        elif isinstance(cache, str):
            with open(cache, "r") as f:
                raw_spec = json.load(f, object_pairs_hook=OrderedDict)
            msg = "cache must be an OrderedDict or str, given {}"
            raise ValueError(msg.format(type(cache)))
        spec = JsonRef.replace_refs(raw_spec)
        classes = parse_api_spec(spec, api_version, resources)
    classes_ = _add_no_underscore_compatibility(classes)
    return classes_

def _add_no_underscore_compatibility(classes):
    """ Add class names without underscores for compatibility.

    Previously, no resources had underscores in APIClient.  Parsing of
    resources has been fixed so now these resources will have underscores.
    This adds an extra class for those resources without an underscore
    for compatibility. Additionally, this removes the resource "feature_flags"
    as APIClient has a name collision with this resource. This will
    be removed in v2.0.0.
    new = ["bocce_clusters", "match_targets", "remote_hosts", "feature_flags"]
    classes_ = {}
    class_names = list(classes.keys())
    for class_name in class_names:
        if class_name != "feature_flags":
            classes_[class_name] = classes[class_name]
        if class_name in new:
            class_name_no_us = "".join(class_name.split("_"))
            classes_[class_name_no_us] = classes[class_name]
    return classes_