Source code for civis.utils._jobs

import logging

from civis import APIClient
from civis.futures import CivisFuture
from civis._deprecation import deprecate_param

log = logging.getLogger(__name__)

[docs]@deprecate_param("v2.0.0", "api_key") def run_job(job_id, api_key=None, client=None, polling_interval=None): """Run a job. Parameters ---------- job_id: str or int The ID of the job. api_key: DEPRECATED str, optional Your Civis API key. If not given, the :envvar:`CIVIS_API_KEY` environment variable will be used. client: :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. polling_interval : int or float, optional The number of seconds between API requests to check whether a result is ready. Returns ------- results: :class:`~civis.futures.CivisFuture` A `CivisFuture` object. """ if client is None: client = APIClient(api_key=api_key) run = return CivisFuture(, (job_id, run["id"]), client=client, polling_interval=polling_interval, poll_on_creation=False, )
[docs]def run_template(id, arguments, JSONValue=False, client=None): """Run a template and return the results. Parameters ---------- id: int The template id to be run. arguments: dict Dictionary of arguments to be passed to the template. JSONValue: bool, optional If True, will return the JSON output of the template. If False, will return the file ids associated with the output results. client: :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. Returns ------- output: dict If JSONValue = False, dictionary of file ids with the keys being their output names. If JSONValue = True, JSON dict containing the results of the template run. Expects only a single JSON result. Will return nothing if either there is no JSON result or there is more than 1 JSON result. Examples -------- >>> # Run template to return file_ids >>> run_template(my_template_id, arguments=my_dict_of_args) {'output': 1234567} >>> # Run template to return JSON output >>> run_template(my_template_id, arguments=my_dict_of_args, JSONValue=True) {'result1': 'aaa', 'result2': 123} """ if client is None: client = APIClient() job = client.scripts.post_custom(id, arguments=arguments) run = client.scripts.post_custom_runs( fut = CivisFuture( client.scripts.get_custom_runs, (,, client=client ) fut.result() outputs = client.scripts.list_custom_runs_outputs(, if JSONValue: json_output = [ o.value for o in outputs if o.object_type == "JSONValue" ] if len(json_output) == 0: log.warning("No JSON output for template {}".format(id)) return if len(json_output) > 1: log.warning( "More than 1 JSON output for template {}" " -- returning only the first one.".format(id) ) # Note that the cast to a dict is to convert # an expected Response object. return dict(json_output[0]) else: file_ids = { o.object_id for o in outputs} return file_ids