osm_common/temporal/activities/lcm.py

#######################################################################################
# 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()