Finally got rid of interfaces!

royalnet/commands/command.py

@@ -3,6 +3,9 @@ from typing import *
 from .commandargs import CommandArgs
 from .commanddata import CommandData
 
+if TYPE_CHECKING:
+    from ..serf import Serf
+
 
 class Command(metaclass=abc.ABCMeta):
     name: str = NotImplemented
@@ -24,8 +27,8 @@ class Command(metaclass=abc.ABCMeta):
     """The syntax of the command, to be displayed when a :py:exc:`InvalidInputError` is raised,
      in the format ``(required_arg) [optional_arg]``."""
 
-    def __init__(self, serf, config):
+    def __init__(self, serf: "Serf", config):
-        self.serf = serf
+        self.serf: "Serf" = serf
         self.config = config
 
     def __str__(self):

royalnet/constellation/constellation.py

@@ -92,9 +92,6 @@ class Constellation:
         self.herald_task: Optional[aio.Task] = None
         """A reference to the :class:`aio.Task` that runs the :class:`rh.Link`."""
 
-        self.Interface: Type[rc.CommandInterface] = self.interface_factory()
-        """The :class:`~rc.CommandInterface` class of this :class:`Constellation`."""
-
         self.events: Dict[str, rc.HeraldEvent] = {}
         """A dictionary containing all :class:`~rc.Event` that can be handled by this :class:`Constellation`."""
 
@@ -149,16 +146,12 @@ class Constellation:
         
         Because of how :mod:`uvicorn` runs, it will stay :const:`None` until the first page is requested."""
 
-    # TODO: is this a good idea?
+    def init_herald(self, herald_cfg: Dict[str, Any]):
-    def interface_factory(self) -> Type[rc.CommandInterface]:
+        """Create a :class:`rh.Link`."""
-        """Create the :class:`rc.CommandInterface` class for the :class:`Constellation`."""
+        herald_cfg["name"] = "constellation"
+        self.herald: rh.Link = rh.Link(rh.Config.from_config(**herald_cfg), self.network_handler)
 
-        # noinspection PyMethodParameters
+    async def call_herald_event(self, destination: str, event_name: str, **kwargs) -> Dict:
-        class GenericInterface(rc.CommandInterface):
-            alchemy: ra.Alchemy = self.alchemy
-            constellation = self
-
-            async def call_herald_event(ci, destination: str, event_name: str, **kwargs) -> Dict:
         """Send a :class:`royalherald.Request` to a specific destination, and wait for a
         :class:`royalherald.Response`."""
         if self.herald is None:
@@ -199,13 +192,6 @@ class Constellation:
             raise rc.ProgramError(f"Other Herald Link returned unknown response:\n"
                                   f"[p]{response}[/p]")
 
-        return GenericInterface
-
-    def init_herald(self, herald_cfg: Dict[str, Any]):
-        """Create a :class:`rh.Link`."""
-        herald_cfg["name"] = "constellation"
-        self.herald: rh.Link = rh.Link(rh.Config.from_config(**herald_cfg), self.network_handler)
-
     async def network_handler(self, message: Union[rh.Request, rh.Broadcast]) -> rh.Response:
         try:
             event: rc.HeraldEvent = self.events[message.handler]
@@ -230,11 +216,9 @@ class Constellation:
 
     def register_events(self, events: List[Type[rc.HeraldEvent]], pack_cfg: Dict[str, Any]):
         for SelectedEvent in events:
-            # Create a new interface
-            interface = self.Interface(config=pack_cfg)
             # Initialize the event
             try:
-                event = SelectedEvent(interface)
+                event = SelectedEvent(serf=self, config=pack_cfg)
             except Exception as e:
                 log.error(f"Skipping: "
                           f"{SelectedEvent.__qualname__} - {e.__class__.__qualname__} in the initialization.")
@@ -266,7 +250,7 @@ class Constellation:
         for SelectedPageStar in page_stars:
             log.debug(f"Registering: {SelectedPageStar.path} -> {SelectedPageStar.__qualname__}")
             try:
-                page_star_instance = SelectedPageStar(interface=self.Interface(pack_cfg))
+                page_star_instance = SelectedPageStar(constellation=self, config=pack_cfg)
             except Exception as e:
                 log.error(f"Skipping: "
                           f"{SelectedPageStar.__qualname__} - {e.__class__.__qualname__} in the initialization.")
@@ -277,7 +261,6 @@ class Constellation:
 
     def run_blocking(self):
         log.info(f"Running Constellation on https://{self.address}:{self.port}/...")
-        loop: aio.AbstractEventLoop = aio.get_event_loop()
         self.running = True
         try:
             uvicorn.run(self.starlette, host=self.address, port=self.port, log_config=UVICORN_LOGGING_CONFIG)

royalnet/constellation/star.py

@@ -1,7 +1,6 @@
 from typing import *
 from starlette.requests import Request
 from starlette.responses import Response
-from royalnet.commands import CommandInterface
 
 if TYPE_CHECKING:
     from .constellation import Constellation
@@ -11,8 +10,9 @@ class Star:
     """A Star is a class representing a part of the website.
 
     It shouldn't be used directly: please use :class:`PageStar` and :class:`ExceptionStar` instead!"""
-    def __init__(self, interface: CommandInterface):
+    def __init__(self, constellation: "Constellation", config):
-        self.interface: CommandInterface = interface
+        self.constellation: "Constellation" = constellation
+        self.config = config
 
     async def page(self, request: Request) -> Response:
         """The function generating the :class:`~starlette.Response` to a web :class:`~starlette.Request`.
@@ -20,31 +20,21 @@ class Star:
         If it raises an error, the corresponding :class:`ExceptionStar` will be used to handle the request instead."""
         raise NotImplementedError()
 
-    @property
-    def constellation(self) -> "Constellation":
-        """A shortcut for the :class:`Constellation`."""
-        return self.interface.constellation
-
     @property
     def alchemy(self):
         """A shortcut for the :class:`~royalnet.alchemy.Alchemy` of the :class:`Constellation`."""
-        return self.interface.constellation.alchemy
+        return self.constellation.alchemy
 
     # noinspection PyPep8Naming
     @property
     def Session(self):
         """A shortcut for the :class:`~royalnet.alchemy.Alchemy` :class:`Session` of the :class:`Constellation`."""
-        return self.interface.constellation.alchemy.Session
+        return self.constellation.alchemy.Session
 
     @property
     def session_acm(self):
         """A shortcut for :func:`.alchemy.session_acm` of the :class:`Constellation`."""
-        return self.interface.constellation.alchemy.session_acm
+        return self.constellation.alchemy.session_acm
-
-    @property
-    def config(self) -> Dict[str, Any]:
-        """A shortcut for the Pack configuration of the :class:`Constellation`."""
-        return self.interface.config
 
     def __repr__(self):
         return f"<{self.__class__.__qualname__}>"

royalnet/serf/matrix/matrixserf.py

@@ -15,6 +15,7 @@ log = logging.getLogger(__name__)
 class MatrixSerf(Serf):
     """A serf that connects to `Matrix <https://matrix.org/>`_ as an user."""
     interface_name = "matrix"
+    prefix = "!"
 
     _identity_table = rb.tables.Matrix
     _identity_column = "matrix_id"
@@ -47,26 +48,14 @@ class MatrixSerf(Serf):
 
         self.Data: Type[rc.CommandData] = self.data_factory()
 
-    def interface_factory(self) -> Type[rc.CommandInterface]:
-        # noinspection PyPep8Naming
-        GenericInterface = super().interface_factory()
-
-        # noinspection PyMethodParameters,PyAbstractClass
-        class DiscordInterface(GenericInterface):
-            name = self.interface_name
-            prefix = "!"
-
-        return DiscordInterface
-
     def data_factory(self) -> Type[rc.CommandData]:
         # noinspection PyMethodParameters,PyAbstractClass
         class MatrixData(rc.CommandData):
             def __init__(data,
-                         interface: rc.CommandInterface,
+                         command: rc.Command,
-                         loop: aio.AbstractEventLoop,
                          room: nio.MatrixRoom,
                          event: nio.Event):
-                super().__init__(interface=interface, loop=loop)
+                super().__init__(command=command)
                 data.room: nio.MatrixRoom = room
                 data.event: nio.Event = event
 
@@ -118,7 +107,8 @@ class MatrixSerf(Serf):
         else:
             session = None
         # Prepare data
-        data = self.Data(interface=command.interface, loop=self.loop, room=room, event=event)
+        # noinspection PyArgumentList
+        data = self.Data(command=command, room=room, event=event)
         # Call the command
         await self.call(command, data, parameters)
         # Close the alchemy session

royalnet/serf/serf.py

@@ -4,7 +4,7 @@ import asyncio as aio
 import sys
 from typing import *
 from sqlalchemy.schema import Table
-from royalnet.commands import *
+import royalnet.commands as rc
 import royalnet.utils as ru
 import royalnet.alchemy as ra
 import royalnet.backpack.tables as rbt
@@ -12,7 +12,6 @@ import royalnet.herald as rh
 import traceback
 import abc
 
-
 log = logging.getLogger(__name__)
 
 
@@ -89,10 +88,10 @@ class Serf(abc.ABC):
         self.herald_task: Optional[aio.Task] = None
         """A reference to the :class:`asyncio.Task` that runs the :class:`Link`."""
 
-        self.events: Dict[str, HeraldEvent] = {}
+        self.events: Dict[str, rc.HeraldEvent] = {}
         """A dictionary containing all :class:`Event` that can be handled by this :class:`Serf`."""
 
-        self.commands: Dict[str, Command] = {}
+        self.commands: Dict[str, rc.Command] = {}
         """The :class:`dict` connecting each command name to its :class:`Command` object."""
 
         for pack_name in packs:
@@ -140,44 +139,44 @@ class Serf(abc.ABC):
         """Send a :class:`royalherald.Request` to a specific destination, and wait for a
         :class:`royalherald.Response`."""
         if self.herald is None:
-            raise UnsupportedError("`royalherald` is not enabled on this serf.")
+            raise rc.UnsupportedError("`royalherald` is not enabled on this serf.")
         request: rh.Request = rh.Request(handler=event_name, data=kwargs)
         response: rh.Response = await self.herald.request(destination=destination, request=request)
         if isinstance(response, rh.ResponseFailure):
             if response.name == "no_event":
-                raise ProgramError(f"There is no event named {event_name} in {destination}.")
+                raise rc.ProgramError(f"There is no event named {event_name} in {destination}.")
             elif response.name == "error_in_event":
                 if response.extra_info["type"] == "CommandError":
-                    raise CommandError(response.extra_info["message"])
+                    raise rc.CommandError(response.extra_info["message"])
                 elif response.extra_info["type"] == "UserError":
-                    raise UserError(response.extra_info["message"])
+                    raise rc.UserError(response.extra_info["message"])
                 elif response.extra_info["type"] == "InvalidInputError":
-                    raise InvalidInputError(response.extra_info["message"])
+                    raise rc.InvalidInputError(response.extra_info["message"])
                 elif response.extra_info["type"] == "UnsupportedError":
-                    raise UnsupportedError(response.extra_info["message"])
+                    raise rc.UnsupportedError(response.extra_info["message"])
                 elif response.extra_info["type"] == "ConfigurationError":
-                    raise ConfigurationError(response.extra_info["message"])
+                    raise rc.ConfigurationError(response.extra_info["message"])
                 elif response.extra_info["type"] == "ExternalError":
-                    raise ExternalError(response.extra_info["message"])
+                    raise rc.ExternalError(response.extra_info["message"])
                 else:
-                    raise ProgramError(f"Invalid error in Herald event '{event_name}':\n"
+                    raise rc.ProgramError(f"Invalid error in Herald event '{event_name}':\n"
                                           f"[b]{response.extra_info['type']}[/b]\n"
                                           f"{response.extra_info['message']}")
             elif response.name == "unhandled_exception_in_event":
-                raise ProgramError(f"Unhandled exception in Herald event '{event_name}':\n"
+                raise rc.ProgramError(f"Unhandled exception in Herald event '{event_name}':\n"
                                       f"[b]{response.extra_info['type']}[/b]\n"
                                       f"{response.extra_info['message']}")
             else:
-                raise ProgramError(f"Unknown response in Herald event '{event_name}':\n"
+                raise rc.ProgramError(f"Unknown response in Herald event '{event_name}':\n"
                                       f"[b]{response.name}[/b]"
                                       f"[p]{response}[/p]")
         elif isinstance(response, rh.ResponseSuccess):
             return response.data
         else:
-            raise ProgramError(f"Other Herald Link returned unknown response:\n"
+            raise rc.ProgramError(f"Other Herald Link returned unknown response:\n"
                                   f"[p]{response}[/p]")
 
-    def register_commands(self, commands: List[Type[Command]], pack_cfg: Dict[str, Any]) -> None:
+    def register_commands(self, commands: List[Type[rc.Command]], pack_cfg: Dict[str, Any]) -> None:
         """Initialize and register all commands passed as argument."""
         # Instantiate the Commands
         for SelectedCommand in commands:
@@ -211,7 +210,7 @@ class Serf(abc.ABC):
         herald_cfg["name"] = self.interface_name
         self.herald: rh.Link = rh.Link(rh.Config.from_config(**herald_cfg), self.network_handler)
 
-    def register_events(self, events: List[Type[HeraldEvent]], pack_cfg: Dict[str, Any]):
+    def register_events(self, events: List[Type[rc.HeraldEvent]], pack_cfg: Dict[str, Any]):
         for SelectedEvent in events:
             # Initialize the event
             try:
@@ -230,7 +229,7 @@ class Serf(abc.ABC):
 
     async def network_handler(self, message: Union[rh.Request, rh.Broadcast]) -> rh.Response:
         try:
-            event: HeraldEvent = self.events[message.handler]
+            event: rc.HeraldEvent = self.events[message.handler]
         except KeyError:
             log.warning(f"No event for '{message.handler}'")
             return rh.ResponseFailure("no_event", f"This serf does not have any event for {message.handler}.")
@@ -239,7 +238,7 @@ class Serf(abc.ABC):
             try:
                 response_data = await event.run(**message.data)
                 return rh.ResponseSuccess(data=response_data)
-            except CommandError as e:
+            except rc.CommandError as e:
                 return rh.ResponseFailure("error_in_event",
                                           f"The event '{message.handler}' raised a {e.__class__.__qualname__}.",
                                           extra_info={
@@ -258,25 +257,25 @@ class Serf(abc.ABC):
         elif isinstance(message, rh.Broadcast):
             await event.run(**message.data)
 
-    async def call(self, command: Command, data: CommandData, parameters: List[str]):
+    async def call(self, command: rc.Command, data: rc.CommandData, parameters: List[str]):
         log.info(f"Calling command: {command.name}")
         try:
             # Run the command
-            await command.run(CommandArgs(parameters), data)
+            await command.run(rc.CommandArgs(parameters), data)
-        except InvalidInputError as e:
+        except rc.InvalidInputError as e:
             await data.reply(f"⚠️ {e.message}\n"
                              f"Syntax: [c]{self.prefix}{command.name} {command.syntax}[/c]")
-        except UserError as e:
+        except rc.UserError as e:
             await data.reply(f"⚠️ {e.message}")
-        except UnsupportedError as e:
+        except rc.UnsupportedError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ExternalError as e:
+        except rc.ExternalError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ConfigurationError as e:
+        except rc.ConfigurationError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ProgramError as e:
+        except rc.ProgramError as e:
             await data.reply(f"⛔️ {e.message}")
-        except CommandError as e:
+        except rc.CommandError as e:
             await data.reply(f"⚠️ {e.message}")
         except Exception as e:
             ru.sentry_exc(e)
@@ -284,23 +283,23 @@ class Serf(abc.ABC):
         finally:
             await data.session_close()
 
-    async def press(self, key: KeyboardKey, data: CommandData):
+    async def press(self, key: rc.KeyboardKey, data: rc.CommandData):
         log.info(f"Calling key_callback: {repr(key)}")
         try:
             await key.press(data)
-        except InvalidInputError as e:
+        except rc.InvalidInputError as e:
             await data.reply(f"⚠️ {e.message}")
-        except UserError as e:
+        except rc.UserError as e:
             await data.reply(f"⚠️ {e.message}")
-        except UnsupportedError as e:
+        except rc.UnsupportedError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ExternalError as e:
+        except rc.ExternalError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ConfigurationError as e:
+        except rc.ConfigurationError as e:
             await data.reply(f"⚠️ {e.message}")
-        except ProgramError as e:
+        except rc.ProgramError as e:
             await data.reply(f"⛔️ {e.message}")
-        except CommandError as e:
+        except rc.CommandError as e:
             await data.reply(f"⚠️ {e.message}")
         except Exception as e:
             ru.sentry_exc(e)

royalnet/serf/telegram/telegramserf.py

@@ -114,7 +114,7 @@ class TelegramSerf(Serf):
                                     parse_mode="HTML",
                                     disable_web_page_preview=True)
 
-            async def reply_image(data, image: BinaryIO, caption: Optional[str] = None) -> None:
+            async def reply_image(data, image: "BinaryIO", caption: Optional[str] = None) -> None:
                 await self.api_call(data.message.chat.send_photo,
                                     photo=image,
                                     caption=escape(caption) if caption is not None else None,