1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-26 21:14:19 +00:00

💥 Implement more blueprints

This commit is contained in:
Steffo 2020-12-11 12:42:41 +01:00
parent c5d1b96f6a
commit abcf89782b
6 changed files with 176 additions and 52 deletions

View file

@ -8,6 +8,7 @@
---------------------------------------------- ----------------------------------------------
.. automodule:: royalnet.engineer.blueprints .. automodule:: royalnet.engineer.blueprints
:imported-members:
``teleporter`` - Function parameter validation ``teleporter`` - Function parameter validation

View file

@ -1 +1,4 @@
from .blueprint import *
from .message import * from .message import *
from .channel import *
from .user import *

View file

@ -0,0 +1,91 @@
import abc
from .. import exc
class Blueprint(metaclass=abc.ABCMeta):
"""
A class containing methods common between all blueprints.
To extend a blueprint, inherit from it while using the :class:`abc.ABCMeta` metaclass, and make all new functions
return :exc:`.exc.NeverAvailableError`:
.. code-block::
class Channel(Blueprint, metaclass=abc.ABCMeta):
def name(self):
raise exc.NeverAvailableError()
To implement a blueprint for a specific chat platform, inherit from the blueprint, override :meth:`__init__`,
:meth:`__hash__` and the methods that are implemented by the platform in question, either returning the
corresponding value or raising :exc:`.exc.NotAvailableError` if there is no data available.
.. code-block::
class ExampleChannel(Channel):
def __init__(self, chat_id: int):
self.chat_id: int = chat_id
def __hash__(self):
return self.chat_id
def name(self):
return ExampleClient.expensive_get_channel_name(self.chat_id)
.. note:: To improve performance, you might want to wrap all data methods in :func:`functools.lru_cache` decorators.
.. code-block::
@functools.lru_cache(24)
def name(self):
return ExampleClient.expensive_get_channel_name(self.chat_id)
"""
@abc.abstractmethod
def __init__(self):
"""
:return: The created object.
"""
raise NotImplementedError()
@abc.abstractmethod
def __hash__(self):
"""
:return: A value that uniquely identifies the channel inside this Python process.
"""
raise NotImplementedError()
def requires(self, *fields) -> None:
"""
Ensure that this blueprint has the specified fields, re-raising the highest priority exception raised between
all of them.
.. code-block::
def print_msg(message: Message):
message.requires(Message.text, Message.timestamp)
print(f"{message.timestamp().isoformat()}: {message.text()}")
:raises .exc.NeverAvailableError: If at least one of the fields raised a :exc:`.exc.NeverAvailableError`.
:raises .exc.NotAvailableError: If no field raised a :exc:`.exc.NeverAvailableError`, but at least one raised a
:exc:`.exc.NotAvailableError`.
"""
exceptions = []
for field in fields:
try:
field(self)
except exc.NeverAvailableError as ex:
exceptions.append(ex)
except exc.NotAvailableError as ex:
exceptions.append(ex)
if len(exceptions) > 0:
raise max(exceptions, key=lambda e: e.priority)
__all__ = (
"Blueprint",
)

View file

@ -0,0 +1,35 @@
from __future__ import annotations
from royalnet.royaltyping import *
import abc
from .. import exc
from .blueprint import Blueprint
class Channel(Blueprint, metaclass=abc.ABCMeta):
"""
An abstract class representing a channel where messages can be sent.
.. seealso:: :class:`.Blueprint`
"""
def name(self) -> str:
"""
:return: The name of the message channel, such as the chat title.
:raises .exc.NeverAvailableError: If the chat platform does not support channel names.
:raises .exc.NotAvailableError: If this channel does not have any name.
"""
raise exc.NeverAvailableError()
def topic(self) -> str:
"""
:return: The topic or description of the message channel.
:raises .exc.NeverAvailableError: If the chat platform does not support channel topics / descriptions.
:raises .exc.NotAvailableError: If this channel does not have any name.
"""
raise exc.NeverAvailableError()
__all__ = (
"Channel",
)

View file

@ -2,64 +2,19 @@ from __future__ import annotations
from royalnet.royaltyping import * from royalnet.royaltyping import *
import abc import abc
import datetime import datetime
import functools
from .. import exc from .. import exc
from .blueprint import Blueprint
from .channel import Channel
class Message(metaclass=abc.ABCMeta): class Message(Blueprint, metaclass=abc.ABCMeta):
""" """
An abstract class representing a generic chat message sent in any platform. An abstract class representing a chat message sent in any platform.
To implement it for a specific platform, override :meth:`__hash__` and the methods returning information that the .. seealso:: :class:`.Blueprint`
platform sometimes provides, either returning the value or raising :exc:`.exc.NotAvailableError`.
All properties are cached using :func:`functools.lru_cache`, so that if they are successful, they are executed only
one time.
""" """
@abc.abstractmethod
def __hash__(self):
"""
:return: A value that uniquely identifies the message inside this Python process.
"""
raise NotImplementedError()
def requires(self, *fields) -> None:
"""
Ensure that this message has the specified fields, raising the highest priority exception between all the
fields.
.. code-block::
def print_msg(message: Message):
message.requires(Message.text, Message.timestamp)
print(f"{message.timestamp().isoformat()}: {message.text()}")
:raises .exc.NeverAvailableError: If at least one of the fields raised a :exc:`.exc.NeverAvailableError`.
:raises .exc.NotAvailableError: If no field raised a :exc:`.exc.NeverAvailableError`, but at least one raised a
:exc:`.exc.NotAvailableError`.
"""
exceptions = []
for field in fields:
try:
field(self)
except exc.NeverAvailableError as ex:
exceptions.append(ex)
except exc.NotAvailableError as ex:
exceptions.append(ex)
if len(exceptions) > 0:
raise max(exceptions, key=lambda e: e.priority)
cache_size = 24
"""
The size of the various :func:`functools.lru_cache`.
"""
@functools.lru_cache(cache_size)
def text(self) -> str: def text(self) -> str:
""" """
:return: The raw text contents of the message. :return: The raw text contents of the message.
@ -68,7 +23,6 @@ class Message(metaclass=abc.ABCMeta):
""" """
raise exc.NeverAvailableError() raise exc.NeverAvailableError()
@functools.lru_cache(cache_size)
def timestamp(self) -> datetime.datetime: def timestamp(self) -> datetime.datetime:
""" """
:return: The :class:`datetime.datetime` at which the message was sent / received. :return: The :class:`datetime.datetime` at which the message was sent / received.
@ -77,7 +31,6 @@ class Message(metaclass=abc.ABCMeta):
""" """
raise exc.NeverAvailableError() raise exc.NeverAvailableError()
@functools.lru_cache(cache_size)
def reply_to(self) -> Message: def reply_to(self) -> Message:
""" """
:return: The :class:`.Message` this message is a reply to. :return: The :class:`.Message` this message is a reply to.
@ -86,6 +39,13 @@ class Message(metaclass=abc.ABCMeta):
""" """
raise exc.NeverAvailableError() raise exc.NeverAvailableError()
def channel(self) -> Channel:
"""
:return: The :class:`.Channel` this message was sent in.
:raises .exc.NeverAvailableError: If the chat platform does not support channels.
:raises .exc.NotAvailableError: If this message was not sent in any channel.
"""
__all__ = ( __all__ = (
"Message", "Message",

View file

@ -0,0 +1,34 @@
from __future__ import annotations
from royalnet.royaltyping import *
import abc
import sqlalchemy.orm
from .. import exc
from .blueprint import Blueprint
class User(Blueprint, metaclass=abc.ABCMeta):
"""
An abstract class representing a chat user.
.. seealso:: :class:`.Blueprint`
"""
def name(self) -> str:
"""
:return: The user's name.
:raises .exc.NeverAvailableError: If the chat platform does not support usernames.
:raises .exc.NotAvailableError: If this user does not have any name.
"""
raise exc.NeverAvailableError()
def database(self, session: sqlalchemy.orm.Session) -> Any:
"""
:param session: A :class:`sqlalchemy.orm.Session` instance to use to fetch the database entry.
:return: The database entry for this user.
"""
__all__ = (
"User",
)