azure/aria/aria-extension-cloudify/src/aria/aria/orchestrator/context/workflow.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 Workflow context.
  18 """
  19
  20 import threading
  21 from contextlib import contextmanager
  22
  23 from .exceptions import ContextException
  24 from .common import BaseContext
  25
  26
  27 class WorkflowContext(BaseContext):
  28     """
  29     Context used during workflow creation and execution.
  30     """
  31     def __init__(self,
  32                  workflow_name,
  33                  parameters=None,
  34                  task_max_attempts=1,
  35                  task_retry_interval=0,
  36                  task_ignore_failure=False,
  37                  *args, **kwargs):
  38         super(WorkflowContext, self).__init__(*args, **kwargs)
  39         self._workflow_name = workflow_name
  40         self._parameters = parameters or {}
  41         self._task_max_attempts = task_max_attempts
  42         self._task_retry_interval = task_retry_interval
  43         self._task_ignore_failure = task_ignore_failure
  44         self._execution_graph = None
  45         self._register_logger()
  46
  47     def __repr__(self):
  48         return (
  49             '{name}(deployment_id={self._service_id}, '
  50             'workflow_name={self._workflow_name}, execution_id={self._execution_id})'.format(
  51                 name=self.__class__.__name__, self=self))
  52
  53     @property
  54     def workflow_name(self):
  55         return self._workflow_name
  56
  57     @property
  58     def execution(self):
  59         """
  60         Execution model.
  61         """
  62         return self.model.execution.get(self._execution_id)
  63
  64     @execution.setter
  65     def execution(self, value):
  66         """
  67         Stores the execution in the storage model API ("MAPI").
  68         """
  69         self.model.execution.put(value)
  70
  71     @property
  72     def node_templates(self):
  73         """
  74         Iterates over nodes templates.
  75         """
  76         key = 'service_{0}'.format(self.model.node_template.model_cls.name_column_name())
  77
  78         return self.model.node_template.iter(
  79             filters={
  80                 key: getattr(self.service, self.service.name_column_name())
  81             }
  82         )
  83
  84     @property
  85     def nodes(self):
  86         """
  87         Iterates over nodes.
  88         """
  89         key = 'service_{0}'.format(self.model.node.model_cls.name_column_name())
  90         return self.model.node.iter(
  91             filters={
  92                 key: getattr(self.service, self.service.name_column_name())
  93             }
  94         )
  95
  96     @property
  97     @contextmanager
  98     def persist_changes(self):
  99         yield
 100         self._model.execution.update(self.execution)
 101
 102
 103 class _CurrentContext(threading.local):
 104     """
 105     Provides a thread-level context, with sugar for the task MAPI.
 106     """
 107
 108     def __init__(self):
 109         super(_CurrentContext, self).__init__()
 110         self._workflow_context = None
 111
 112     def _set(self, value):
 113         self._workflow_context = value
 114
 115     def get(self):
 116         """
 117         Retrieves the current workflow context.
 118         """
 119         if self._workflow_context is not None:
 120             return self._workflow_context
 121         raise ContextException("No context was set")
 122
 123     @contextmanager
 124     def push(self, workflow_context):
 125         """
 126         Switches the current context to the provided context.
 127         """
 128         prev_workflow_context = self._workflow_context
 129         self._set(workflow_context)
 130         try:
 131             yield self
 132         finally:
 133             self._set(prev_workflow_context)
 134
 135 current = _CurrentContext()