osm_common/temporal/activities/lcm.py

   1 #######################################################################################
   2 # Copyright ETSI Contributors and Others.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # 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
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #######################################################################################
  17
  18 from dataclasses import dataclass
  19
  20 from osm_common.dbbase import DbBase
  21 from osm_common.temporal.activities.base import BaseActivity
  22 from osm_common.temporal.states import LcmOperationState
  23
  24
  25 class NsLcmNoOp(BaseActivity):
  26     """
  27     This is a simple No Operation Activity that simply logs the data
  28     with which it was called. It can be used as a placeholder when
  29     developing workflows, or can be enhanced with logic to throw
  30     exceptions on specific conditions to test exception handling in
  31     a workflow.
  32     """
  33
  34     @dataclass
  35     class Input:
  36         """
  37         Input dataclass for workflows that run as LCM operations.
  38
  39         Attributes:
  40         -----------
  41
  42         nslcmop: dict
  43             A dictionary representing the nslcmop record from the
  44             database.
  45         """
  46
  47         nslcmop: dict
  48
  49     async def __call__(self, activity_input: Input) -> None:
  50         raise NotImplementedError()
  51
  52
  53 class UpdateNsLcmOperationState(BaseActivity):
  54     """
  55     Changes the state of a LCM operation task.  Should be done to
  56     indicate progress, or completion of the task itself.
  57
  58     Collaborators:
  59         DB Write:           nslcmops
  60
  61     Raises  (Retryable):
  62         DbException         If the target DB record does not exist or DB is not reachable.
  63
  64     Activity Lifecycle:
  65         This activity will not report a heartbeat due to its
  66         short-running nature.
  67         As this is a direct DB update, it is not recommended to have
  68         any specific retry policy
  69     """
  70
  71     @dataclass
  72     class Input:
  73         """
  74         Input dataclass for updating LCM Operations in the Mongo nslcmops
  75         collection.  The following attributes will be updated automatically
  76           - statusEnteredTime
  77           - _admin.modified
  78
  79         Attributes:
  80         -----------
  81         op_id: str
  82             The operation (task) id for this activity.  This is the key
  83             to the record in nslcmops collection that will be updated.
  84
  85         op_state : LcmOperationState
  86             A representation of the state of the specified operation id,
  87             such as PROCESSING, COMPLETED, or FAILED.
  88
  89         stage: str
  90             Human readable checkpoint message, intended only to give the
  91             user feedback.
  92
  93         error_message: str
  94             Human readable error message if any failure occurred.
  95
  96         detailed_status : str
  97             Human readable message providing additional details to the
  98             operation state, such as the error message explaining why
  99             the operation failed.
 100         """
 101
 102         op_id: str
 103         op_state: LcmOperationState
 104         stage: str
 105         error_message: str
 106         detailed_status: str
 107
 108     def __init__(self, db: DbBase):
 109         super().__init__()
 110         self.db: DbBase = db
 111
 112     async def __call__(self, activity_input: Input) -> None:
 113         raise NotImplementedError()