[python][plan][api] Init Event, Action, Function and WorkflowPlan.

Author: wenjin272

.gitignore

@@ -7,3 +7,4 @@ python/flink_agents/flink_agents.egg-info/
 python/build/
 python/dist
 python/.DS_Store
+__pycache__

python/flink_agents/api/event.py

@@ -0,0 +1,40 @@
+from abc import ABC
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+
+
+class Event(BaseModel, ABC, extra="allow"):
+    """Base class for all event types in the system.
+
+    Attributes:
+    ----------
+    id : UUID
+        Unique identifier for the event, automatically generated using uuid4.
+    """
+
+    id: UUID = Field(default_factory=uuid4)
+
+
+class InputEvent(Event):
+    """Event generated by the framework, carrying an input data that
+    arrives at the workflow.
+    """
+
+    input: str
+
+
+class OutputEvent(Event):
+    """Event representing a result from workflow. By generating an OutputEvent,
+    actions can emit output data.
+
+    Attributes:
+    ----------
+    isLegal : bool
+        Is the result legal or not.
+    result : str
+        The final result returned by the workflow.
+    """
+
+    isLegal: bool
+    result: str

python/flink_agents/plan/action.py

@@ -0,0 +1,26 @@
+from typing import List, Type
+
+from pydantic import BaseModel
+from python.flink_agents.api.event import Event
+from python.flink_agents.plan.function import Function
+
+
+class Action(BaseModel):
+    """Representation of a workflow action with event listening and function execution.
+
+    This class encapsulates a named workflow action that listens for specific event
+    types and executes an associated function when those events occur.
+
+    Attributes:
+    ----------
+    name : str
+        Name/identifier of the workflow Action.
+    exec : Function
+        To be executed when the Action is triggered.
+    listen_event_types : List[Type[Event]]
+        List of event types that will trigger this Action's execution.
+    """
+
+    name: str
+    exec: Function
+    listen_event_types: List[Type[Event]]

python/flink_agents/plan/function.py

@@ -0,0 +1,84 @@
+import importlib
+import inspect
+from typing import Any, Callable, Dict, Tuple
+
+from pydantic import BaseModel
+
+
+class Function(BaseModel):
+    """Descriptor for a callable function, storing module and qualified name for dynamic
+    retrieval.
+
+    This class allows serialization and lazy loading of functions by storing their
+    module and
+    qualified name. The actual callable is loaded on-demand when the instance is called.
+
+    Attributes:
+    ----------
+    module : str
+        Name of the Python module where the function is defined.
+    qualname : str
+        Qualified name of the function (e.g., 'ClassName.method' for class methods).
+    __func: Callable
+        Internal cache for the resolved function
+    """
+
+    module: str
+    qualname: str
+    __func: Callable = None
+
+    @staticmethod
+    def from_callable(func: Callable) -> "Function":
+        """Create a Function descriptor from an existing callable.
+
+        Parameters
+        ----------
+        func : Callable
+            The function or method to be wrapped.
+
+        Returns:
+        -------
+        Function
+            A Function instance with module and qualname populated based on the input
+            callable.
+        """
+        return Function(
+            module=inspect.getmodule(func).__name__,
+            qualname=func.__qualname__,
+            __func=func,
+        )
+
+    def __call__(self, *args: Tuple[Any, ...], **kwargs: Dict[str, Any]) -> Any:
+        """Execute the stored function with provided arguments.
+
+        Lazily loads the function from its module and qualified name if not already
+        cached.
+
+        Parameters
+        ----------
+        *args : tuple
+            Positional arguments to pass to the function.
+        **kwargs : dict
+            Keyword arguments to pass to the function.
+
+        Returns:
+        -------
+        Any
+            The result of calling the resolved function with the provided arguments.
+
+        Notes:
+        -----
+        If the function is a method (qualified name contains a class reference), it will
+        resolve the method from the corresponding class.
+        """
+        if self.__func is None:
+            module = importlib.import_module(self.module)
+            if "." in self.qualname:
+                # Handle class methods (e.g., 'ClassName.method')
+                classname, methodname = self.qualname.rsplit(".", 1)
+                clazz = getattr(module, classname)
+                self.__func = getattr(clazz, methodname)
+            else:
+                # Handle standalone functions
+                self.__func = getattr(module, self.qualname)
+        return self.__func(*args, **kwargs)

python/flink_agents/plan/tests/__init__.py

@@ -0,0 +1,33 @@
+from python.flink_agents.api.event import InputEvent, OutputEvent
+from python.flink_agents.plan.action import Action
+from python.flink_agents.plan.function import Function
+from python.flink_agents.plan.workflow_plan import WorkflowPlan
+
+
+def increment(event: InputEvent) -> OutputEvent: # noqa: D103
+    value = event.input
+    value = int(value) + 1
+    return OutputEvent(isLegal=True, result=str(value))
+
+
+def test_simplest_workflow_plan() -> None: # noqa: D103
+    INCREMENT_ACTION = Action(
+        name="increment",
+        exec=Function.from_callable(increment),
+        listen_event_types=[InputEvent],
+    )
+    actions = {InputEvent: [INCREMENT_ACTION]}
+    workflow_plan = WorkflowPlan(actions=actions)
+
+    input_event = InputEvent(input="1")
+    input_event_triggered_actions = workflow_plan.get_action(type(input_event))
+    for action in input_event_triggered_actions:
+        result_event = action.exec(input_event)
+        if isinstance(result_event, OutputEvent):
+            assert result_event.isLegal
+            print(result_event.result)
+            assert result_event.result == "2"
+
+
+if __name__ == "__main__":
+    test_simplest_workflow_plan()

python/flink_agents/plan/tests/test_workflow_plan.py

@@ -0,0 +1,32 @@
+from typing import Dict, List, Type
+
+from pydantic import BaseModel
+from python.flink_agents.api.event import Event
+from python.flink_agents.plan.action import Action
+
+
+class WorkflowPlan(BaseModel):
+    """Workflow plan compiled from user defined workflow.
+
+    Attributes:
+    ----------
+    actions : Dict[Type[Event], List[Action]]
+        Mapping of event types to the list of Actions that listen to them.
+    """
+
+    actions: Dict[Type[Event], List[Action]]
+
+    def get_action(self, event_type: Type[Event]) -> List[Action]:
+        """Get steps that listen to the specified event type.
+
+        Parameters
+        ----------
+        event_type : Type[Event]
+            The event type to query.
+
+        Returns:
+        -------
+        list[Action]
+            List of Actions that will respond to this event type.
+        """
+        return self.actions[event_type]

python/flink_agents/plan/workflow_plan.py

@@ -13,3 +13,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 build
+pydantic==2.11.4