azure/aria/aria-extension-cloudify/src/aria/aria/orchestrator/context/operation.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 """
  17 Operation contexts.
  18 """
  19
  20 import threading
  21 from contextlib import contextmanager
  22
  23 import aria
  24 from aria.utils import file
  25 from . import common
  26
  27
  28 class BaseOperationContext(common.BaseContext):
  29     """
  30     Base class for contexts used during operation creation and execution.
  31     """
  32
  33     def __init__(self, task_id, actor_id, **kwargs):
  34         self._task_id = task_id
  35         self._actor_id = actor_id
  36         self._thread_local = threading.local()
  37         self._destroy_session = kwargs.pop('destroy_session', False)
  38         logger_level = kwargs.pop('logger_level', None)
  39         super(BaseOperationContext, self).__init__(**kwargs)
  40         self._register_logger(task_id=self.task.id, level=logger_level)
  41
  42     def __repr__(self):
  43         details = 'function={task.function}; ' \
  44                   'operation_arguments={task.arguments}'\
  45             .format(task=self.task)
  46         return '{name}({0})'.format(details, name=self.name)
  47
  48     @property
  49     def task(self):
  50         """
  51         The task in the model storage.
  52         """
  53         # SQLAlchemy prevents from accessing an object which was created on a different thread.
  54         # So we retrieve the object from the storage if the current thread isn't the same as the
  55         # original thread.
  56
  57         if not hasattr(self._thread_local, 'task'):
  58             self._thread_local.task = self.model.task.get(self._task_id)
  59         return self._thread_local.task
  60
  61     @property
  62     def plugin_workdir(self):
  63         """
  64         A work directory that is unique to the plugin and the service ID.
  65         """
  66         if self.task.plugin is None:
  67             return None
  68         plugin_workdir = '{0}/plugins/{1}/{2}'.format(self._workdir,
  69                                                       self.service.id,
  70                                                       self.task.plugin.name)
  71         file.makedirs(plugin_workdir)
  72         return plugin_workdir
  73
  74     @property
  75     def serialization_dict(self):
  76         context_dict = {
  77             'name': self.name,
  78             'service_id': self._service_id,
  79             'task_id': self._task_id,
  80             'actor_id': self._actor_id,
  81             'workdir': self._workdir,
  82             'model_storage': self.model.serialization_dict if self.model else None,
  83             'resource_storage': self.resource.serialization_dict if self.resource else None,
  84             'execution_id': self._execution_id,
  85             'logger_level': self.logger.level
  86         }
  87         return {
  88             'context_cls': self.__class__,
  89             'context': context_dict
  90         }
  91
  92     @classmethod
  93     def instantiate_from_dict(cls, model_storage=None, resource_storage=None, **kwargs):
  94         if model_storage:
  95             model_storage = aria.application_model_storage(**model_storage)
  96         if resource_storage:
  97             resource_storage = aria.application_resource_storage(**resource_storage)
  98
  99         return cls(model_storage=model_storage,
 100                    resource_storage=resource_storage,
 101                    destroy_session=True,
 102                    **kwargs)
 103
 104     def close(self):
 105         if self._destroy_session:
 106             self.model.log._session.remove()
 107             self.model.log._engine.dispose()
 108
 109     @property
 110     @contextmanager
 111     def persist_changes(self):
 112         yield
 113         self.model.task.update(self.task)
 114
 115
 116 class NodeOperationContext(BaseOperationContext):
 117     """
 118     Context for node operations.
 119     """
 120
 121     @property
 122     def node(self):
 123         """
 124         The node of the current operation.
 125         """
 126         return self.model.node.get(self._actor_id)
 127
 128     @property
 129     def node_template(self):
 130         """
 131         The node template of the current operation.
 132         """
 133         return self.node.node_template
 134
 135
 136 class RelationshipOperationContext(BaseOperationContext):
 137     """
 138     Context for relationship operations.
 139     """
 140
 141     @property
 142     def relationship(self):
 143         """
 144         The relationship instance of the current operation.
 145         """
 146         return self.model.relationship.get(self._actor_id)
 147
 148     @property
 149     def source_node(self):
 150         """
 151         The relationship source node.
 152         """
 153         return self.relationship.source_node
 154
 155     @property
 156     def source_node_template(self):
 157         """
 158         The relationship source node template.
 159         """
 160         return self.source_node.node_template
 161
 162     @property
 163     def target_node(self):
 164         """
 165         The relationship target node.
 166         """
 167         return self.relationship.target_node
 168
 169     @property
 170     def target_node_template(self):
 171         """
 172         The relationship target node template.
 173         """
 174         return self.target_node.node_template