Source code for cloudify_rest_client.plugins_update

import warnings

from cloudify_rest_client import utils
from cloudify_rest_client.exceptions import CloudifyClientError
from cloudify_rest_client.responses import ListResponse

[docs] class PluginsUpdate(dict): def __init__(self, update): self.update(update) @property def id(self): return self['id'] @property def state(self): return self['state'] @property def blueprint_id(self): return self['blueprint_id'] @property def temp_blueprint_id(self): return self['temp_blueprint_id'] @property def execution_id(self): return self['execution_id'] @property def deployments_to_update(self): return self['deployments_to_update'] @property def deployments_per_tenant(self): return self['deployments_per_tenant'] @property def created_at(self): return self['created_at'] @property def forced(self): return self['forced'] @property def tenant_name(self): return self['tenant_name']
[docs] class PluginsUpdateClient(object): """ Cloudify's plugins update management client. """ def __init__(self, api): self.api = api self._uri_prefix = 'plugins-updates' self._wrapper_cls = PluginsUpdate
[docs] def get(self, plugins_update_id, _include=None, **kwargs): """ Gets a plugins update by its id. :param plugins_update_id: PluginsUpdate's id to get. :param _include: List of fields to include in response. :return: The plugins update details. """ assert plugins_update_id uri = '/{self._uri_prefix}/{id}'.format( self=self, id=plugins_update_id) response = self.api.get(uri, _include=_include, params=kwargs) return self._wrapper_cls(response)
def _wrap_list(self, response): return ListResponse( [self._wrapper_cls(item) for item in response['items']], response['metadata'] )
[docs] def list(self, _include=None, sort=None, is_descending=False, **kwargs): """ Returns a list of available plugins updates. :param _include: List of fields to include in response. :param sort: Key for sorting the list. :param is_descending: True for descending order, False for ascending. :param kwargs: Optional filter fields. For a list of available fields see the REST service's models.PluginsUpdate.fields :return: Plugins list. """ params = kwargs if sort: params['_sort'] = '-' + sort if is_descending else sort response = self.api.get('/{self._uri_prefix}'.format(self=self), _include=_include, params=params) return self._wrap_list(response)
[docs] def inject(self, blueprint_id, force=False, created_by=None, created_at=None, execution_id=None, state=None, update_id=None, affected_deployments=None, deployments_per_tenant=None, all_tenants=None, temp_blueprint_id=None): return PluginsUpdate( '/{self._uri_prefix}/{}/update/initiate'.format(blueprint_id, self=self), data=_data_from_kwargs( force=force, created_by=created_by, created_at=created_at, execution_id=execution_id, state=state, update_id=update_id, all_tenants=all_tenants, affected_deployments=affected_deployments, deployments_per_tenant=deployments_per_tenant, temp_blueprint_id=temp_blueprint_id, ), ))
[docs] def update_plugins(self, blueprint_id, force=False, plugin_names=None, to_latest=None, all_to_latest=True, to_minor=None, all_to_minor=False, mapping=None, auto_correct_types=False, reevaluate_active_statuses=False, all_tenants=False): """ Updates the plugins in all the deployments that use the given blueprint. :param blueprint_id: blueprint ID to perform the update with. :param force: if to forcefully update when other non-active plugins updates exists associated with this blueprint. :param plugin_names: list of plugin names to update (only those). :param to_latest: list of plugin names to be upgraded to the latest installed version. :param all_to_latest: update all (selected) plugins to the latest installed version of a plugin. :param to_minor: list of plugin names to be upgraded to the latest installed minor version (i.e. major versions of the plugin in use and the upgraded one will match) :param all_to_minor: update all (selected) plugins to the latest installed minor version :param mapping: detailed information on required plugin update (overrides all other arguments/settings concerning version constraints) :param auto_correct_types: auto_correct_types flag to run deployments update with :param reevaluate_active_statuses: reevaluate active plugin-updates' and deployment-updates' states based on relevant executions statuses. :param all_tenants: defines if plugin update process should update any deployments based on blueprint_id (owned by any tenant) :return: a PluginUpdate object. """ if mapping and mapping.get('updates'): warnings.warn("The 'mapping file' was used during the update; " "remember to update your blueprint files", RuntimeWarning) else: mapping = {} response = '/{self._uri_prefix}/{}/update/initiate'.format(blueprint_id, self=self), data=_data_from_kwargs( plugin_names=plugin_names, to_latest=to_latest, all_to_latest=all_to_latest, to_minor=to_minor, all_to_minor=all_to_minor, mapping=mapping, force=force, auto_correct_types=auto_correct_types, reevaluate_active_statuses=reevaluate_active_statuses, all_tenants=all_tenants ) ) return PluginsUpdate(response)
[docs] def finalize_plugins_update(self, plugins_update_id): """ Finalize the plugins update (for internal use). :return: a PluginUpdate object. """ response = '/{self._uri_prefix}/{}/update/finalize'.format(plugins_update_id, self=self) ) return PluginsUpdate(response)
[docs] def dump(self, plugins_update_ids=None): """Generate plugins updates' attributes for a snapshot. :param plugins_update_ids: A list of plugins updates' identifiers, if not empty, used to select specific plugins updates to be dumped. :returns: A generator of dictionaries, which describe plugins updates' attributes. """ entities = utils.get_all( self.api.get, f'/{self._uri_prefix}', params={'_get_data': True}, _include=['id', 'state', 'forced', 'all_tenants', 'blueprint_id', 'execution_id', 'created_by', 'created_at', 'deployments_to_update', 'deployments_per_tenant', 'temp_blueprint_id'], ) if not plugins_update_ids: return entities return (e for e in entities if e['id'] in plugins_update_ids)
[docs] def restore(self, entities, logger): """Restore plugins updates from a snapshot. :param entities: An iterable (e.g. a list) of dictionaries describing plugins updates to be restored. :param logger: A logger instance. """ for entity in entities: entity['update_id'] = entity.pop('id') entity['affected_deployments'] = entity.pop( 'deployments_to_update', None) entity['force'] = entity.pop('forced', None) try: self.inject(**entity) except CloudifyClientError as exc: logger.error("Error restoring plugins update " f"{entity['update_id']}: {exc}")
def _data_from_kwargs(**kwargs): data = {} for k, v in kwargs.items(): if v: data[k] = v return data or None