OSMENG-1155 Implementation of Constants and Dataclasses
Add stubs for common elements, workflows and activities
Change-Id: If7a2aa8a6e6627df5293154bf48da742dea57e1c
Signed-off-by: Dario Faccin <dario.faccin@canonical.com>
diff --git a/osm_common/temporal/activities/lcm.py b/osm_common/temporal/activities/lcm.py
new file mode 100644
index 0000000..9298bd1
--- /dev/null
+++ b/osm_common/temporal/activities/lcm.py
@@ -0,0 +1,113 @@
+#######################################################################################
+# Copyright ETSI Contributors and Others.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#######################################################################################
+
+from dataclasses import dataclass
+
+from osm_common.dbbase import DbBase
+from osm_common.temporal.activities.base import BaseActivity
+from osm_common.temporal.states import LcmOperationState
+
+
+class NsLcmNoOp(BaseActivity):
+ """
+ This is a simple No Operation Activity that simply logs the data
+ with which it was called. It can be used as a placeholder when
+ developing workflows, or can be enhanced with logic to throw
+ exceptions on specific conditions to test exception handling in
+ a workflow.
+ """
+
+ @dataclass
+ class Input:
+ """
+ Input dataclass for workflows that run as LCM operations.
+
+ Attributes:
+ -----------
+
+ nslcmop: dict
+ A dictionary representing the nslcmop record from the
+ database.
+ """
+
+ nslcmop: dict
+
+ async def __call__(self, activity_input: Input) -> None:
+ raise NotImplementedError()
+
+
+class UpdateNsLcmOperationState(BaseActivity):
+ """
+ Changes the state of a LCM operation task. Should be done to
+ indicate progress, or completion of the task itself.
+
+ Collaborators:
+ DB Write: nslcmops
+
+ Raises (Retryable):
+ DbException If the target DB record does not exist or DB is not reachable.
+
+ Activity Lifecycle:
+ This activity will not report a heartbeat due to its
+ short-running nature.
+ As this is a direct DB update, it is not recommended to have
+ any specific retry policy
+ """
+
+ @dataclass
+ class Input:
+ """
+ Input dataclass for updating LCM Operations in the Mongo nslcmops
+ collection. The following attributes will be updated automatically
+ - statusEnteredTime
+ - _admin.modified
+
+ Attributes:
+ -----------
+ op_id: str
+ The operation (task) id for this activity. This is the key
+ to the record in nslcmops collection that will be updated.
+
+ op_state : LcmOperationState
+ A representation of the state of the specified operation id,
+ such as PROCESSING, COMPLETED, or FAILED.
+
+ stage: str
+ Human readable checkpoint message, intended only to give the
+ user feedback.
+
+ error_message: str
+ Human readable error message if any failure occurred.
+
+ detailed_status : str
+ Human readable message providing additional details to the
+ operation state, such as the error message explaining why
+ the operation failed.
+ """
+
+ op_id: str
+ op_state: LcmOperationState
+ stage: str
+ error_message: str
+ detailed_status: str
+
+ def __init__(self, db: DbBase):
+ super().__init__()
+ self.db: DbBase = db
+
+ async def __call__(self, activity_input: Input) -> None:
+ raise NotImplementedError()