from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, Union
from .contents import Content
from .responses import Message, Response
[docs]class Action(ABC):
"""A generic action. Every action must inherit from this class and override its methods."""
__slots__ = ()
[docs] @abstractmethod
def process(self, payload: Optional[dict] = None) -> Union[Message, Response]:
"""Process the payload and return a response.
Args:
payload: An incoming message payload.
Returns:
A message or a response object.
"""
pass
[docs] @abstractmethod
def serialize(self) -> dict:
"""Serialize the class instance to a dictionary.
Returns:
A serialized class instance.
"""
pass
[docs]class MessageAction(Action):
"""An action that returns a message.
Args:
text: The message text.
"""
__slots__ = 'text',
def __init__(self, text: str) -> None:
self.text = text
def __repr__(self) -> str:
return f'MessageAction({repr(self.text)})'
[docs] def process(self, payload: Optional[dict] = None) -> Message:
return Message(Content(text=self.text), payload=payload)
[docs] def serialize(self) -> dict:
return {
'text': self.text,
}
[docs]class GoBackAction(Action):
"""An action that returns user to the one of previous menus.
Args:
count: How many times to go back.
"""
__slots__ = 'count',
def __init__(self, count: int = 1) -> None:
self.count = count
def __repr__(self) -> str:
return f'GoBackAction({self.count})'
[docs] def process(self, payload: Optional[dict] = None) -> Response:
return Response(go_back_count=self.count)
[docs] def serialize(self) -> dict:
res = {}
if self.count != 1:
res['count'] = self.count
return res
[docs]class ExecuteAction(Action):
"""An action that executes a code and optionally returns a message with the result.
Args:
command: The code to execute.
return_text: Whether to return result as a message with text.
"""
__slots__ = 'command', 'return_text'
def __init__(self, command: str, return_text: bool = False) -> None:
self.command = command
self.return_text = return_text
def __repr__(self) -> str:
return f'ExecuteAction({repr(self.command)}, {self.return_text})'
[docs] def process(self, payload: Optional[dict] = None) -> Optional[Message]:
if self.return_text:
return Message(Content(text=str(eval(self.command))), payload=payload)
else:
exec(self.command)
[docs] def serialize(self) -> dict:
res = {
'command': self.command,
}
if self.return_text is not False:
res['return_text'] = self.return_text
return res
[docs]class FunctionAction(Action):
"""An action that runs a function and optionally returns a response.
Args:
function: The function to run.
templates: The response templates.
"""
__slots__ = 'function', 'templates'
def __init__(self, function, templates: Optional[Dict[Any, Union[Message, Response]]] = None) -> None:
if templates is None:
templates = {}
self.function = function
self.templates = templates
def __repr__(self) -> str:
return f'FunctionAction({self.function}, {self.templates})'
[docs] def process(self, payload: Optional[dict] = None) -> Union[Message, Response]:
if payload is None:
payload = {}
func_res = self.function(payload)
if func_res in self.templates:
res = self.templates[func_res]
res.update_payload(payload)
return res
else:
func_res.update_payload(payload)
return func_res
[docs] def serialize(self) -> dict:
res = {
'function': self.function.__name__,
}
if self.templates:
res['templates'] = [{
'case': k,
'type': v.__class__.__name__,
**v.serialize(),
} for k, v in self.templates.items()]
return res