Source code for kappa_sdk.user_tasks.context

from typing import Optional, Dict, Any
import json




class Context:
    """User task context.

    Provides an ability to persist values between different runs of the User Task.

    .. automethod:: __getitem__
    .. automethod:: __setitem__
    """

    def __init__(self, field_id: Optional[str] = None, well_id: Optional[str] = None, context_dict: Optional[Dict[str, Any]] = None, workflow_id: Optional[str] = None) -> None:
        self.__field_id: Optional[str] = field_id
        self.__well_id: Optional[str] = well_id
        self.__workflow_id: Optional[str] = workflow_id
        if context_dict is None:
            context_dict = dict()
        self._context_dict: Dict[str, Any] = context_dict

    @property
    def field_id(self) -> Optional[str]:
        """ Gets the field id.
        """
        return self.__field_id

    @field_id.setter
    def field_id(self, value: str) -> None:
        self.__field_id = value

    @property
    def well_id(self) -> Optional[str]:
        """ Gets the well id.
        """
        return self.__well_id

    @well_id.setter
    def well_id(self, value: str) -> None:
        self.__well_id = value

    @property
    def workflow_id(self) -> Optional[str]:
        """ Gets the workflow id.
        """
        return self.__workflow_id

    @workflow_id.setter
    def workflow_id(self, value: str) -> None:
        self.__workflow_id = value



    def __getitem__(self, key: str) -> Optional[Any]:
        """Gets the value associated with the specified string key.

        If the value could not be found, a default one is returned.
        Only string keys are supported.

        Parameters
        ----------
        key:
            The name of the key.

        Raises
        ------
        TypeError
            If key has an unexpected type.

        """
        if not isinstance(key, str):
            raise TypeError('Only string keys are supported in the user task context, but key of type {} was given instead.'.format(type(key)))
        return self._context_dict.copy().get(key)




    def __setitem__(self, key: str, value: Any) -> None:
        """Sets the value for the specified string key.

        Only string keys are supported.

        Parameters
        ----------
        key:
            The name of the key.
        value:
            The value to save.

        Raises
        ------
        TypeError
            If key has an unexpected type.

        """
        if not isinstance(key, str):
            raise TypeError('Only string keys are supported in the user task context, but key of type {} was given instead.'.format(type(key)))
        self._context_dict[key] = value


    def dumps(self) -> str:
        return json.dumps(self._context_dict, default=str)