Trixy7/trixy_core/scheduler/action/callback.py

# -*- coding: utf-8 -*-
"""
Callback Action - Führt Python-Funktionen aus.
"""

import asyncio
import inspect
from typing import Any, Callable, Awaitable

from trixy_core.scheduler.action.base import Action


class CallbackAction(Action):
    """
    Action die eine Python-Funktion aufruft.

    Beispiel:
        - CallbackAction(my_function)
        - CallbackAction(async_function, pass_context=True)
        - CallbackAction(lambda: print("Hello"))
    """

    def __init__(
        self,
        callback: Callable[..., Any] | Callable[..., Awaitable[Any]],
        args: tuple = (),
        kwargs: dict | None = None,
        pass_context: bool = False,
        name: str | None = None,
        timeout_seconds: float = 60.0,
        retry_on_failure: bool = False,
        max_retries: int = 3,
    ):
        """
        Initialisiert die Action.

        Args:
            callback: Funktion (sync oder async)
            args: Positionale Argumente
            kwargs: Keyword-Argumente
            pass_context: Kontext als erstes Argument übergeben
            name: Action-Name
            timeout_seconds: Timeout
            retry_on_failure: Bei Fehler wiederholen
            max_retries: Maximale Wiederholungen
        """
        func_name = getattr(callback, "__name__", "callback")
        super().__init__(
            name or f"callback_{func_name}",
            timeout_seconds,
            retry_on_failure,
            max_retries,
        )
        self._callback = callback
        self._args = args
        self._kwargs = kwargs or {}
        self._pass_context = pass_context
        self._is_async = asyncio.iscoroutinefunction(callback)

    @property
    def is_async(self) -> bool:
        """Prüft ob Callback async ist."""
        return self._is_async

    @property
    def pass_context(self) -> bool:
        """Prüft ob Kontext übergeben wird."""
        return self._pass_context

    async def _execute(self, context: dict) -> Any:
        """Führt den Callback aus."""
        # Argumente vorbereiten
        args = self._args
        kwargs = dict(self._kwargs)

        if self._pass_context:
            args = (context,) + args

        # Callback ausführen
        if self._is_async:
            result = await asyncio.wait_for(
                self._callback(*args, **kwargs),
                timeout=self._timeout_seconds,
            )
        else:
            # Sync-Funktion in Thread ausführen
            loop = asyncio.get_event_loop()
            result = await asyncio.wait_for(
                loop.run_in_executor(
                    None,
                    lambda: self._callback(*args, **kwargs),
                ),
                timeout=self._timeout_seconds,
            )

        return result

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "callback": getattr(self._callback, "__name__", "<lambda>"),
            "pass_context": self._pass_context,
            # args/kwargs können nicht immer serialisiert werden
        }

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "Callback-Funktion"

    def __repr__(self) -> str:
        func_name = getattr(self._callback, "__name__", "<lambda>")
        return f"CallbackAction(callback={func_name})"


class MethodAction(Action):
    """
    Action die eine Methode auf einem Objekt aufruft.

    Beispiel:
        - MethodAction(service, "refresh")
        - MethodAction(manager, "cleanup", kwargs={"force": True})
    """

    def __init__(
        self,
        obj: Any,
        method_name: str,
        args: tuple = (),
        kwargs: dict | None = None,
        name: str | None = None,
        timeout_seconds: float = 60.0,
    ):
        """
        Initialisiert die Action.

        Args:
            obj: Objekt auf dem Methode aufgerufen wird
            method_name: Name der Methode
            args: Positionale Argumente
            kwargs: Keyword-Argumente
            name: Action-Name
            timeout_seconds: Timeout
        """
        super().__init__(name or f"method_{method_name}", timeout_seconds)
        self._obj = obj
        self._method_name = method_name
        self._args = args
        self._kwargs = kwargs or {}

    @property
    def method_name(self) -> str:
        """Methoden-Name."""
        return self._method_name

    async def _execute(self, context: dict) -> Any:
        """Führt die Methode aus."""
        # Methode holen
        method = getattr(self._obj, self._method_name)

        if not callable(method):
            raise ValueError(f"{self._method_name} ist nicht aufrufbar")

        # Methode ausführen
        if asyncio.iscoroutinefunction(method):
            result = await asyncio.wait_for(
                method(*self._args, **self._kwargs),
                timeout=self._timeout_seconds,
            )
        else:
            loop = asyncio.get_event_loop()
            result = await asyncio.wait_for(
                loop.run_in_executor(
                    None,
                    lambda: method(*self._args, **self._kwargs),
                ),
                timeout=self._timeout_seconds,
            )

        return result

    @classmethod
    def display_name(cls) -> str:
        """Anzeigename fuer die TUI."""
        return "Methoden-Aufruf"

    def to_dict(self) -> dict:
        """Konvertiert zu Dictionary."""
        return {
            **self._base_dict(),
            "object_type": type(self._obj).__name__,
            "method_name": self._method_name,
        }