Source code for civis.ml._helper

from collections import namedtuple
import logging

from civis import APIClient
from civis.ml._model import _get_template_ids_all_versions

__all__ = ['list_models', 'put_models_shares_groups',
           'put_models_shares_users', 'delete_models_shares_groups',
           'delete_models_shares_users']
log = logging.getLogger(__name__)

# sentinel value for default author value
SENTINEL = namedtuple('Sentinel', [])()


[docs]def list_models(job_type="train", author=SENTINEL, client=None, **kwargs): """List a user's CivisML models. Parameters ---------- job_type : {"train", "predict", None} The type of model job to list. If "train", list training jobs only (including registered models trained outside of CivisML). If "predict", list prediction jobs only. If None, list both. author : int, optional User id of the user whose models you want to list. Defaults to the current user. Use ``None`` to list models from all users. client : :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. **kwargs : kwargs Extra keyword arguments passed to `client.scripts.list_custom()` See Also -------- APIClient.scripts.list_custom """ if job_type == "train": job_types = ('training',) elif job_type == "predict": job_types = ('prediction',) elif job_type is None: job_types = ('training', 'prediction') else: raise ValueError("Parameter 'job_type' must be None, 'train', " "or 'predict'.") if client is None: client = APIClient() template_id_list = [ ids[_job_type] for _job_type in job_types for ids in _get_template_ids_all_versions(client).values() ] # Applying set() because we don't want repeated IDs # between the version-less production alias and the versioned alias. template_id_str = ', '.join(str(tmp) for tmp in set(template_id_list)) if author is SENTINEL: author = client.users.list_me().id # default to showing most recent models first kwargs.setdefault('order_dir', 'desc') models = client.scripts.list_custom(from_template_id=template_id_str, author=author, **kwargs) return models
[docs]def put_models_shares_users(id, user_ids, permission_level, client=None, share_email_body='DEFAULT', send_shared_email='DEFAULT'): """Set the permissions users have on this object Use this on both training and scoring jobs. If used on a training job, note that "read" permission is sufficient to score the model. Parameters ---------- id : integer The ID of the resource that is shared. user_ids : list An array of one or more user IDs. permission_level : string Options are: "read", "write", or "manage". client : :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. share_email_body : string, optional Custom body text for e-mail sent on a share. send_shared_email : boolean, optional Send email to the recipients of a share. Returns ------- readers : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string writers : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string owners : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string total_user_shares : integer For owners, the number of total users shared. For writers and readers, the number of visible users shared. total_group_shares : integer For owners, the number of total groups shared. For writers and readers, the number of visible groups shared. """ kwargs = {} if send_shared_email != 'DEFAULT': kwargs['send_shared_email'] = send_shared_email if share_email_body != 'DEFAULT': kwargs['share_email_body'] = share_email_body return _share_model(id, user_ids, permission_level, entity_type='users', client=client, **kwargs)
[docs]def put_models_shares_groups(id, group_ids, permission_level, client=None, share_email_body='DEFAULT', send_shared_email='DEFAULT'): """Set the permissions groups have on this model. Use this on both training and scoring jobs. If used on a training job, note that "read" permission is sufficient to score the model. Parameters ---------- id : integer The ID of the resource that is shared. group_ids : list An array of one or more group IDs. permission_level : string Options are: "read", "write", or "manage". client : :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. share_email_body : string, optional Custom body text for e-mail sent on a share. send_shared_email : boolean, optional Send email to the recipients of a share. Returns ------- readers : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string writers : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string owners : dict:: - users : list:: - id : integer - name : string - groups : list:: - id : integer - name : string total_user_shares : integer For owners, the number of total users shared. For writers and readers, the number of visible users shared. total_group_shares : integer For owners, the number of total groups shared. For writers and readers, the number of visible groups shared. """ kwargs = {} if send_shared_email != 'DEFAULT': kwargs['send_shared_email'] = send_shared_email if share_email_body != 'DEFAULT': kwargs['share_email_body'] = share_email_body return _share_model(id, group_ids, permission_level, entity_type='groups', client=client, **kwargs)
def _share_model(job_id, entity_ids, permission_level, entity_type, client=None, **kwargs): """Share a container job and all run outputs with requested entities""" client = client or APIClient() if entity_type not in ['groups', 'users']: raise ValueError("'entity_type' must be one of ['groups', 'users']. " "Got '{0}'.".format(entity_type)) log.debug("Sharing object %d with %s %s at permission level %s.", job_id, entity_type, entity_ids, permission_level) _func = getattr(client.scripts, "put_containers_shares_" + entity_type) result = _func(job_id, entity_ids, permission_level, **kwargs) # CivisML relies on several run outputs attached to each model run. # Go through and share all outputs on each run. runs = client.scripts.list_containers_runs(job_id, iterator=True) for run in runs: log.debug("Sharing outputs on %d, run %s.", job_id, run.id) outputs = client.scripts.list_containers_runs_outputs(job_id, run.id) for _output in outputs: if _output['object_type'] == 'File': _func = getattr(client.files, "put_shares_" + entity_type) obj_permission = permission_level elif _output['object_type'] == 'Project': _func = getattr(client.projects, "put_shares_" + entity_type) if permission_level == 'read': # Users must be able to add to projects to use the model obj_permission = 'write' else: obj_permission = permission_level elif _output['object_type'] == 'JSONValue': _func = getattr(client.json_values, "put_shares_" + entity_type) obj_permission = permission_level else: log.debug("Found a run output of type %s, ID %s; not sharing " "it.", _output['object_type'], _output['object_id']) continue _oid = _output['object_id'] # Don't send share emails for any of the run outputs. _func(_oid, entity_ids, obj_permission, send_shared_email=False) return result
[docs]def delete_models_shares_users(id, user_id, client=None): """Revoke the permissions a user has on this object Use this function on both training and scoring jobs. Parameters ---------- id : integer The ID of the resource that is shared. user_id : integer The ID of the user. client : :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. Returns ------- None Response code 204: success """ return _unshare_model(id, user_id, entity_type='users', client=client)
[docs]def delete_models_shares_groups(id, group_id, client=None): """Revoke the permissions a group has on this object Use this function on both training and scoring jobs. Parameters ---------- id : integer The ID of the resource that is shared. group_id : integer The ID of the group. client : :class:`civis.APIClient`, optional If not provided, an :class:`civis.APIClient` object will be created from the :envvar:`CIVIS_API_KEY`. Returns ------- None Response code 204: success """ return _unshare_model(id, group_id, entity_type='groups', client=client)
def _unshare_model(job_id, entity_id, entity_type, client=None): """Revoke permissions on a container job and all run outputs for the requested entity (singular) """ client = client or APIClient() if entity_type not in ['groups', 'users']: raise ValueError("'entity_type' must be one of ['groups', 'users']. " "Got '{0}'.".format(entity_type)) log.debug("Revoking permissions on object %d for %s %s.", job_id, entity_type, entity_id) _func = getattr(client.scripts, "delete_containers_shares_" + entity_type) result = _func(job_id, entity_id) # CivisML relies on several run outputs attached to each model run. # Go through and revoke permissions for outputs on each run. runs = client.scripts.list_containers_runs(job_id, iterator=True) endpoint_name = "delete_shares_" + entity_type for run in runs: log.debug("Unsharing outputs on %d, run %s.", job_id, run.id) outputs = client.scripts.list_containers_runs_outputs(job_id, run.id) for _output in outputs: if _output['object_type'] == 'File': _func = getattr(client.files, endpoint_name) elif _output['object_type'] == 'Project': _func = getattr(client.projects, endpoint_name) elif _output['object_type'] == 'JSONValue': _func = getattr(client.json_values, endpoint_name) else: log.debug("Found run output of type %s, ID %s; not unsharing " "it.", _output['object_type'], _output['object_id']) continue _func(_output['object_id'], entity_id) return result