Index

class Action:
    """
    Object that models an action.

    These actions are unbound and return a bound action when accessed using the owning object.
    """

    __slots__ = [
        "_schema_validator",
        "argument_schema",
        "create_task",
        "idempotent",
        "isclassmethod",
        "iscoroutine",
        "isparameterized",
        "obj",
        "owner",
        "request_as_argument",
        "return_value_schema",
        "safe",
        "state",
        "synchronous",
    ]

    state: tuple[Enum | str] | None
    """state machine state(s) in which this action can be executed, any state when None"""

    def __init__(self, obj: MethodType) -> None:
        """
        Initialize an Action.

        Parameters
        ----------
        obj: MethodType
            the method that is being wrapped as an action
        """
        self.obj = obj
        self.state = None
        self.iscoroutine = False
        self.isclassmethod = False
        self.isparameterized = False
        self.request_as_argument = False
        self.create_task = False
        self.safe = False
        self.idempotent = False
        self.synchronous = True
        self.argument_schema = None
        self.return_value_schema = None
        self._schema_validator = None

    def __set_name__(self, owner, name):
        self.owner = owner

    def __str__(self) -> str:
        return f"<Action({self.owner.__name__}.{self.obj.__name__})>"

    def __eq__(self, other) -> bool:
        if not isinstance(other, Action):
            return False
        return self.obj == other.obj

    def __hash__(self) -> int:
        return hash(self.obj)

    def __get__(self, instance, owner):
        if instance is None and not self.isclassmethod:
            return self
        if self.iscoroutine:
            return BoundAsyncAction(self.obj, self, instance, owner)
        return BoundSyncAction(self.obj, self, instance, owner)

    def __call__(self, *args, **kwargs):
        raise NotImplementedError(
            f"Cannot invoke unbound action {self.name} of {self.owner.__name__}."
            + " Bound methods must be called, not the action itself. Use the appropriate instance to call the method."
        )

    @property
    def name(self) -> str:
        """Name of the action."""
        return self.obj.__name__

    @property
    def schema_validator(self) -> BaseSchemaValidator | None:
        """
        Validator for the arguments of this action, None if the action has no validation.

        Built from `argument_schema` the first time it is needed, which is when the action is first invoked.
        Further calls return the cached instance.
        """
        if self._schema_validator is None and self.argument_schema:
            self._schema_validator = SchemaValidators.for_schema(self.argument_schema)(self.argument_schema)
        return self._schema_validator

    def to_metadata(self, owner_inst: Thing | ThingMeta | None = None, format: str = "wot") -> ActionMetadata:
        """
        Generates a `ActionAffordance` TD fragment for this Action.

        Parameters
        ----------
        owner_inst: Thing, optional
            The instance of the owning `Thing` object. If not supplied, the class is used.

        Returns
        -------
        ActionAffordance
            the affordance TD fragment for this action
        """
        from hololinked import MetadataFormats

        return MetadataFormats.get(format).action.from_descriptor(
            self,
            owner_inst or self.owner,
        )

class BoundAction:
    """A bound action, base class for both sync and async methods."""

    obj: FunctionType | MethodType

    __slots__ = [
        "action",
        "bound_obj",
        "obj",
        "owner",
        "owner_inst",
    ]

    def __init__(self, obj: FunctionType | MethodType, descriptor: Action, owner_inst, owner) -> None:
        self.obj = obj
        self.action = descriptor
        self.owner = owner
        self.owner_inst = owner_inst
        self.bound_obj = owner if descriptor.isclassmethod else owner_inst

    @property
    def descriptor(self) -> Action:
        """The action descriptor."""
        return self.action

    def __post_init__(self):
        # never called, neither possible to call, only type hinting
        # owner class and instance
        self.owner: ThingMeta
        self.owner_inst: Thing
        self.obj: FunctionType
        self.action: Action

    def validate_call(self, args, kwargs: dict[str, Any]) -> None:
        """
        Validate the call to the action, like payload, state machine state etc.

        Errors are raised as exceptions.

        Parameters
        ----------
        args: tuple
            positional arguments to the action
        kwargs: dict
            keyword arguments to the action

        Raises
        ------
        StateMachineError
            if the action cannot be executed in the current state of the owning thing
        RuntimeError
            if the action explicity accepts only keyword arguments but some positional arguments are given
        """
        if self.action.isparameterized and len(args) > 0:
            raise RuntimeError("parameterized functions cannot have positional arguments")
        if self.owner_inst is None:
            return
        if self.action.state is None or (
            hasattr(self.owner_inst, "state_machine")
            and self.owner_inst.state_machine.current_state in self.action.state  # ty: ignore[unresolved-attribute]
        ):
            if self.action.schema_validator is not None:
                self.action.schema_validator.validate_method_call(args, kwargs)
        else:
            raise StateMachineError(
                f"Thing '{self.owner_inst}' is in '{self.owner_inst.state}' state, however action can be executed only in '{self.action.state}' state"
            )

    @property
    def name(self) -> str:
        """Name of the action."""
        return self.obj.__name__

    def __call__(self, *args, **kwargs):
        raise NotImplementedError("call must be implemented by subclass")

    def external_call(self, *args, **kwargs):
        """
        Validated call to the action with state machine and payload checks.

        Returns
        -------
        Any
            the return value of the action
        """
        raise NotImplementedError("external_call must be implemented by subclass")

    def __str__(self):
        return f"<BoundAction({self.owner.__name__}.{self.obj.__name__} of {self.owner_inst.id})>"

    def __eq__(self, value):
        if not isinstance(value, BoundAction):
            return False
        return self.obj == value.obj

    def __hash__(self):
        return hash(str(self))

    def __getattribute__(self, name):
        # https://docs.python.org/3/howto/descriptor.html#functions-and-methods
        if name == "__doc__":
            return self.obj.__doc__
        return super().__getattribute__(name)

    def to_metadata(self, owner_inst: Thing | ThingMeta | None = None, format: str = "wot") -> ActionMetadata:
        """
        Generates a `ActionAffordance` TD fragment for this Action.

        Parameters
        ----------
        owner_inst: Thing, optional
            The instance of the owning `Thing` object. If not supplied, the class is used.

        Returns
        -------
        ActionAffordance
            the affordance TD fragment for this action
        """
        return Action.to_metadata(self.descriptor, owner_inst or self.owner_inst or self.owner, format=format)

class BoundSyncAction(BoundAction):
    """
    Non-async(io) action call.

    The call is passed to the method as-it-is to allow local
    invocation without state machine checks. Use `external_call` to have validation.
    """

    def external_call(self, *args, **kwargs):
        """
        Validated call to the action with state machine and payload checks.

        Returns
        -------
        Any
            the return value of the action
        """
        self.validate_call(args, kwargs)
        return self.__call__(*args, **kwargs)

    def __call__(self, *args, **kwargs):
        if self.action.isclassmethod:
            return self.obj(*args, **kwargs)
        return self.obj(self.bound_obj, *args, **kwargs)

class BoundAsyncAction(BoundAction):
    """
    async(io) action call.

    The call is passed to the method as-it-is to allow local
    invocation without state machine checks. Use `external_call` to have validation.
    """

    async def external_call(self, *args, **kwargs):
        """
        Validated call to the action with state machine and payload checks.

        Returns
        -------
        Any
            the return value of the action
        """
        self.validate_call(args, kwargs)
        return await self.__call__(*args, **kwargs)

    async def __call__(self, *args, **kwargs):
        if self.action.isclassmethod:
            return await self.obj(*args, **kwargs)
        return await self.obj(self.bound_obj, *args, **kwargs)