API Reference¶

This page is autogenerated from the docstrings inside the code.

Alchemy¶

Relational database classes and methods.

class royalnet.alchemy.Alchemy(database_uri: str, tables: Set)¶

A wrapper around sqlalchemy.orm that allows the instantiation of multiple engines at once while maintaining a single declarative class for all of them.

__init__(database_uri: str, tables: Set)¶

Create a new Alchemy object.

Parameters

database_uri – The database URI .
tables – The set of tables to be created and used in the selected database. Check the tables submodule for more details.

get(table: Union[str, type]) → sqlalchemy.sql.schema.Table¶

Get the table with a specified name or class.

Parameters: table – The table name or table class you want to get.
Raises: TableNotFoundError – if the requested table was not found.

session_acm()¶

Create a Session as a async context manager (that can be used in async with statements).

The Session will be closed safely when the context manager exits (even in case of error).

Example

You can use the async context manager like this:

async with alchemy.session_acm() as session:
    # Do some stuff
    ...
    # Commit the session
    await asyncify(session.commit)

session_cm()¶

Create a Session as a context manager (that can be used in with statements).

The Session will be closed safely when the context manager exits (even in case of error).

Example

You can use the context manager like this:

with alchemy.session_cm() as session:
    # Do some stuff
    ...
    # Commit the session
    session.commit()

royalnet.alchemy.table_dfs(starting_table: sqlalchemy.sql.schema.Table, ending_table: sqlalchemy.sql.schema.Table) → tuple¶

Depth-first-search for the path from the starting table to the ending table.

Returns: A tuple containing the path, starting from the starting table and ending at the ending table.

exception royalnet.alchemy.AlchemyException¶: Base class for Alchemy exceptions.

exception royalnet.alchemy.TableNotFoundException¶: The requested table was not found.

Backpack¶

A Pack that is imported by default by all royalnet instances.

Bard¶

class royalnet.bard.YtdlInfo(info: Dict[str, Any])¶

A wrapper around youtube_dl extracted info.

__init__(info: Dict[str, Any])¶: Create a YtdlInfo from the dict returned by the youtube_dl.YoutubeDL.extract_info() function.

Warning

Does not download the info, to do that use retrieve_for_url().

classmethod from_url(url, loop: Optional[asyncio.events.AbstractEventLoop] = None, **ytdl_args) → List[royalnet.bard.ytdlinfo.YtdlInfo]¶

Fetch the info for an url through youtube_dl.

Returns: A list containing the infos for the requested videos.

class royalnet.bard.YtdlFile(url: str, info: Optional[royalnet.bard.ytdlinfo.YtdlInfo] = None, filename: Optional[str] = None, ytdl_args: Optional[Dict[str, Any]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None)¶

A representation of a file download with youtube_dl.

__init__(url: str, info: Optional[royalnet.bard.ytdlinfo.YtdlInfo] = None, filename: Optional[str] = None, ytdl_args: Optional[Dict[str, Any]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None)¶: Create a YtdlFile instance.

Warning

Please avoid using directly __init__(), use from_url() instead!

aopen()¶

Open the downloaded file as an async context manager (and download it if it isn’t available yet).

Example

You can use the async context manager like this:

async with ytdlfile.aopen() as file:
    b: bytes = file.read()

default_ytdl_args = {'ignoreerrors': True, 'no_warnings': False, 'noplaylist': True, 'outtmpl': '%(epoch)s-%(title)s-%(id)s.%(ext)s', 'quiet': False}¶

async delete_asap()¶: As soon as nothing is using the file, delete it.

async download_file() → None¶: Download the file.

classmethod from_url(url: str, **ytdl_args) → List[royalnet.bard.ytdlfile.YtdlFile]¶: Create a list of YtdlFile from a URL.

property has_info¶: Does the YtdlFile have info available?

property is_downloaded¶: Has the file been downloaded yet?

async retrieve_info() → None¶: Retrieve info about the YtdlFile through youtube_dl.

class royalnet.bard.YtdlMp3(ytdl_file: royalnet.bard.ytdlfile.YtdlFile)¶

A representation of a YtdlFile conversion to mp3.

async convert_to_mp3() → None¶: Convert the file to mp3 with ffmpeg.

async delete_asap() → None¶: Delete the mp3 file.

classmethod from_url(url, **ytdl_args) → List[royalnet.bard.ytdlmp3.YtdlMp3]¶: Create a list of YtdlMp3 from a URL.

property info¶: Shortcut to get the YtdlInfo of the object.

property is_converted¶: Has the file been converted?

exception royalnet.bard.BardError¶: Base class for bard errors.

exception royalnet.bard.YtdlError¶: Base class for errors caused by youtube_dl.

exception royalnet.bard.NotFoundError¶: The requested resource wasn’t found.

exception royalnet.bard.MultipleFilesError¶: The resource contains multiple media files.

Commands¶

class royalnet.commands.CommandInterface¶

property alchemy¶: A shortcut for serf.alchemy.

async call_herald_action(destination: str, event_name: str, args: dict) → dict¶

property loop¶: A shortcut for serf.loop.

name = NotImplemented¶

The name of the CommandInterface that’s being implemented.

Examples

telegram, discord, console…

prefix = NotImplemented¶

The prefix used by commands on the interface.

Examples

/ on Telegram, ! on Discord

register_herald_action(event_name: str, coroutine: Callable[[Any], Awaitable[dict]])¶

serf = NotImplemented¶

A reference to the Serf that is implementing this CommandInterface.

Examples

A reference to a TelegramSerf.

unregister_herald_action(event_name: str)¶

class royalnet.commands.Command(interface: royalnet.commands.commandinterface.CommandInterface)¶

property alchemy¶: A shortcut for interface.alchemy.

aliases = []¶

A list of possible aliases for a command.

Example

To be able to call /e as an alias for /example, one should set aliases to ["e"].

description = NotImplemented¶: A small description of the command, to be displayed when the command is being autocompleted.

property loop¶: A shortcut for interface.loop.

name = NotImplemented¶

The main name of the command.

Example

To be able to call /example on Telegram, the name should be "example".

async run(args: royalnet.commands.commandargs.CommandArgs, data: royalnet.commands.commanddata.CommandData) → None¶

syntax = ''¶: The syntax of the command, to be displayed when a InvalidInputError is raised, in the format (required_arg) [optional_arg].

tables = {}¶: A set of royalnet.alchemy tables that must exist for this command to work.

class royalnet.commands.CommandData(interface: royalnet.commands.commandinterface.CommandInterface, session: Optional[Session], loop: asyncio.events.AbstractEventLoop)¶

async delete_invoking(error_if_unavailable=False) → None¶

Delete the invoking message, if supported by the interface.

The invoking message is the message send by the user that contains the command.

Parameters: error_if_unavailable – if True, raise an exception if the message cannot been deleted.

async get_author(error_if_none: bool = False)¶

Try to find the identifier of the user that sent the message. That probably means, the database row identifying the user.

Parameters: error_if_none – Raise an exception if this is True and the call has no author.

async reply(text: str) → None¶

Send a text message to the channel where the call was made.

Parameters: text – The text to be sent, possibly formatted in the weird undescribed markup that I’m using.

property session¶

Get the Alchemy Session, if it is available.

Raises: UnsupportedError – if no session is available.

async session_commit()¶: Commit the changes to the session.

class royalnet.commands.CommandArgs¶

An interface to easily access the arguments of a command.

Inherits from list.

__getitem__(item)¶

Access arguments as if they were a list.

Raises: royalnet.error.InvalidInputError – if the requested argument does not exist.

Examples

# /pasta spaghetti aldente
>>> self[0]
"spaghetti"
>>> self[1]
"aldente"
>>> self[2]
# InvalidInputError: Missing argument #3.
>>> self[0:2]
["spaghetti", "aldente"]

joined(*, require_at_least=0) → str¶

Get the arguments as a space-joined string.

Parameters

require_at_least – the minimum amount of arguments required, will raise InvalidInputError if the
is not fullfilled. (requirement) –

Raises

royalnet.error.InvalidInputError – if there are less than require_at_least arguments.

Returns

The space-joined string.

Examples

# /pasta spaghetti aldente
>>> self.joined()
"spaghetti aldente"
>>> self.joined(require_at_least=3)
# InvalidInputError: Not enough arguments specified (minimum is 3).

match(pattern: Union[str, Pattern], *flags) → Sequence[AnyStr]¶

Match the joined() string to a re.Pattern-like object.

Parameters: pattern – The regex pattern to be passed to re.match().
Raises: InvalidInputError – if the pattern doesn’t match.
Returns: The matched groups, as returned by re.Match.groups().

optional(index: int, default=None) → Optional[str]¶

Get the argument at a specific index, but don’t raise an error if nothing is found, instead returning the default value.

Parameters

index – The index of the argument you want to retrieve.
default – The value returned if the argument is missing.

Returns

Either the argument or the default value, defaulting to None.

Examples

# /pasta spaghetti aldente
>>> self.optional(0)
"spaghetti"
>>> self.optional(2)
None
>>> self.optional(2, default="carbonara")
"carbonara"

exception royalnet.commands.CommandError(message='')¶

Something went wrong during the execution of this command.

Display an error message to the user, explaining what went wrong.

exception royalnet.commands.InvalidInputError(message='')¶: The command has received invalid input and cannot complete.

exception royalnet.commands.UnsupportedError(message='')¶: A requested feature is not available on this interface.

exception royalnet.commands.ConfigurationError(message='')¶: The command cannot work because of a wrong configuration by part of the Royalnet admin.

exception royalnet.commands.ExternalError(message='')¶: The command failed to execute, but the problem was because of an external factor (such as an external API going down).

exception royalnet.commands.UserError(message='')¶: The command failed to execute, and the error is because of something that the user did.

Constellation¶

The part of royalnet that handles the webserver and webpages.

It uses many features of starlette.

class royalnet.constellation.Constellation(secrets_name: str, database_uri: str, page_stars: List[Type[royalnet.constellation.star.PageStar]] = None, exc_stars: List[Type[royalnet.constellation.star.ExceptionStar]] = None, *, debug: bool = True)¶

The class that represents the webserver.

It runs multiple Star , which represent the routes of the website.

It also handles the Alchemy connection, and it will eventually support Herald connections too.

alchemy = None¶: The Alchemy of this Constellation.

get_secret(username: str) → Optional[str]¶

Get a Royalnet secret from the keyring.

Parameters: username – the name of the secret that should be retrieved.

classmethod run_process(address: str, port: int, secrets_name: str, database_uri: str, page_stars: List[Type[royalnet.constellation.star.PageStar]] = None, exc_stars: List[Type[royalnet.constellation.star.ExceptionStar]] = None, *, debug: bool = True)¶

Blockingly create and run the Constellation.

This should be used as the target of a multiprocessing.Process.

Parameters

address – The IP address this Constellation should bind to.
port – The port this Constellation should listen for requests on.

running = None¶: Is the Constellation currently running?

secrets_name = None¶: The secrets_name this Constellation is currently using.

starlette = None¶: The starlette.Starlette app.

class royalnet.constellation.Star(constellation: Constellation)¶

A Star is a class representing a part of the website.

It shouldn’t be used directly: please use PageStar and ExceptionStar instead!

property Session¶: A shortcut for the alchemy Session of the Constellation.

property alchemy¶: A shortcut for the Alchemy of the Constellation.

async page(request: Request) → Response¶

The function generating the Response to a web Request.

If it raises an error, the corresponding ExceptionStar will be used to handle the request instead.

property session_acm¶: A shortcut for the Alchemy.session_acm() of the Constellation.

tables = {}¶: The set of alchemy table classes required by this Star to function.

class royalnet.constellation.PageStar(constellation: Constellation)¶

A PageStar is a class representing a single route of the website (for example, /api/user/get).

To create a new website route you should create a new class inheriting from this class with a function overriding page() and changing the values of path and optionally methods and tables.

methods = ['GET']¶

The HTTP methods supported by the Star, in form of a list.

By default, a Star only supports the GET method, but more can be added.

Example

methods: List[str] = ["GET", "POST", "PUT", "DELETE"]

path = NotImplemented¶

The route of the star.

Example

path: str = '/api/user/get'

class royalnet.constellation.ExceptionStar(constellation: Constellation)¶

An ExceptionStar is a class that handles an Exception raised by another star by returning a different response than the one originally intended.

The handled exception type is specified in the error.

It can also handle standard webserver errors, such as 404 Not Found: to handle them, set error to an int of the corresponding error code.

To create a new exception handler you should create a new class inheriting from this class with a function overriding page() and changing the values of error and optionally tables.

error = None¶

The error that should be handled by this star. It should be either a subclass of Exception, or the int of an HTTP error code.

Examples

error: int = 404

error: Type[Exception] = ValueError

royalnet.constellation.shoot(code: int, description: str) → starlette.responses.JSONResponse¶: Create a error JSONResponse with the passed error code and description.

Herald¶

class royalnet.herald.Config(name: str, address: str, port: int, secret: str, secure: bool = False, path: str = '/')¶

copy(name: Optional[str] = None, address: Optional[str] = None, port: Optional[int] = None, secret: Optional[str] = None, secure: Optional[bool] = None, path: Optional[str] = None)¶: Create an exact copy of this configuration, but with different parameters.

property url¶

exception royalnet.herald.HeraldError¶: A generic royalnet.herald error.

exception royalnet.herald.ConnectionClosedError¶: The Link’s connection was closed unexpectedly. The link can’t be used anymore.

exception royalnet.herald.LinkError¶: An error for something that happened in a Link.

exception royalnet.herald.InvalidServerResponseError¶: The Server sent invalid data to the Link.

exception royalnet.herald.ServerError¶: An error for something that happened in a Server.

class royalnet.herald.Link(config: royalnet.herald.config.Config, request_handler, *, loop: asyncio.events.AbstractEventLoop = None)¶

async broadcast(destination: str, broadcast: royalnet.herald.broadcast.Broadcast) → None¶

async connect()¶: Connect to the Server at master_uri.

async identify() → None¶

async receive() → royalnet.herald.package.Package¶

Recieve a Package from the Server.

Raises: ConnectionClosedError –

async request(destination: str, request: royalnet.herald.request.Request) → royalnet.herald.response.Response¶

async run()¶: Blockingly run the Link.

async send(package: royalnet.herald.package.Package)¶: Send a package to the Server.

class royalnet.herald.Package(data: dict, *, source: str, destination: str, source_conv_id: Optional[str] = None, destination_conv_id: Optional[str] = None)¶

A data type with which a Link communicates with a Server or another Link.

Contains info about the source and the destination.

__init__(data: dict, *, source: str, destination: str, source_conv_id: Optional[str] = None, destination_conv_id: Optional[str] = None)¶

Create a Package.

Parameters

data – The data that should be sent.
source – The nid of the node that created this Package.
destination – The link_type of the destination node, or alternatively, the nid of the node. Can also be the NULL value to send the message to nobody.
source_conv_id – The conversation id of the node that created this package. Akin to the sequence number on IP packets.
destination_conv_id – The conversation id of the node that this Package is a reply to.

static from_dict(d) → royalnet.herald.package.Package¶: Create a Package from a dictionary.

static from_json_bytes(b: bytes) → royalnet.herald.package.Package¶: Create a Package from UTF-8-encoded JSON bytes.

static from_json_string(string: str) → royalnet.herald.package.Package¶: Create a Package from a JSON string.

reply(data) → royalnet.herald.package.Package¶

Reply to this Package with another Package.

Parameters: data – The data that should be sent. Usually a Request.
Returns: The reply Package.

to_dict() → dict¶: Convert the Package into a dictionary.

to_json_bytes() → bytes¶: Convert the Package into UTF-8-encoded JSON bytes.

to_json_string() → str¶: Convert the Package into a JSON string.

class royalnet.herald.Request(handler: str, data: dict, msg_type: Optional[str] = None)¶

A request sent from a Link to another.

It contains the name of the requested handler, in addition to the data.

classmethod from_dict(d: dict)¶

to_dict()¶

class royalnet.herald.Response¶

A base class to be inherited by all other response types.

classmethod from_dict(d: dict) → royalnet.herald.response.Response¶: Recreate the response from a received dict.

to_dict() → dict¶: Prepare the Response to be sent by converting it to a JSONable dict.

class royalnet.herald.ResponseSuccess(data: Optional[dict] = None)¶: A response to a successful Request.

class royalnet.herald.ResponseFailure(name: str, description: str, extra_info: Optional[dict] = None)¶: A response to a invalid Request.

class royalnet.herald.Server(config: royalnet.herald.config.Config, *, loop: asyncio.events.AbstractEventLoop = None)¶

find_client(*, nid: str = None, link_type: str = None) → List[royalnet.herald.server.ConnectedClient]¶

find_destination(package: royalnet.herald.package.Package) → List[royalnet.herald.server.ConnectedClient]¶

Find a list of destinations for the package.

Parameters: package – The package to find the destination of.
Returns: A list of ConnectedClient to send the package to.

async listener(websocket: websockets.server.WebSocketServerProtocol, path)¶

async route_package(package: royalnet.herald.package.Package) → None¶: Executed every time a package is received and must be routed somewhere.

async run()¶

run_blocking()¶

serve()¶

class royalnet.herald.Broadcast(handler: str, data: dict, msg_type: Optional[str] = None)¶

classmethod from_dict(d: dict)¶

to_dict()¶

Serf¶

class royalnet.serf.Serf(*, alchemy_config: Optional[royalnet.serf.alchemyconfig.AlchemyConfig] = None, commands: List[Type[royalnet.commands.command.Command]] = None, network_config: Optional[royalnet.herald.config.Config] = None, secrets_name: str = '__default__')¶

An abstract class, to be used as base to implement Royalnet bots on multiple interfaces (such as Telegram or Discord).

Data = None¶: The CommandData class of this Serf.

Interface = None¶: The CommandInterface class of this Serf.

alchemy = None¶: The Alchemy object connecting this Serf to the database.

async call(command: royalnet.commands.command.Command, data: royalnet.commands.commanddata.CommandData, parameters: List[str])¶

commands = None¶: The dict connecting each command name to its Command object.

data_factory() → Type[royalnet.commands.commanddata.CommandData]¶: Create the CommandData for the Serf.

static find_tables(alchemy_config: royalnet.serf.alchemyconfig.AlchemyConfig, commands: List[Type[royalnet.commands.command.Command]]) → Set[type]¶

Find the :class:`Table`s required by the Serf.

System Message: WARNING/2 (d:\codice\royalnet\royalnet\serf\serf.py:docstring of royalnet.serf.Serf.find_tables, line 1); backlink

Inline interpreted text or phrase reference start-string without end-string.

Warning

This function will return a wrong result if there are tables between the master table and the identity table that aren’t included by a command.

Returns: A list of :class:`Table`s.

get_secret(username: str)¶

Get a Royalnet secret from the keyring.

Parameters: username – the name of the secret that should be retrieved.

herald = None¶: The Link object connecting the Serf to the rest of the herald network.

herald_handlers = None¶: A dict linking Request event names to coroutines returning a dict that will be sent as Response to the event.

herald_task = None¶: A reference to the asyncio.Task that runs the Link.

init_alchemy(alchemy_config: royalnet.serf.alchemyconfig.AlchemyConfig, tables: Set[type]) → None¶: Create and initialize the Alchemy with the required tables, and find the link between the master table and the identity table.

init_network(config: royalnet.herald.config.Config)¶: Create a Link, and run it as a asyncio.Task.

static init_sentry(dsn)¶

interface_factory() → Type[royalnet.commands.commandinterface.CommandInterface]¶: Create the CommandInterface class for the Serf.

interface_name = NotImplemented¶

loop = None¶: The event loop this Serf is running on.

async network_handler(message: Union[royalnet.herald.request.Request, royalnet.herald.broadcast.Broadcast]) → royalnet.herald.response.Response¶

register_commands(commands: List[Type[royalnet.commands.command.Command]]) → None¶

Initialize and register all commands passed as argument.

If called again during the execution of the bot, all current commands will be replaced with the new ones.

Warning

Hot-replacing commands was never tested and probably doesn’t work.

async run()¶: A coroutine that starts the event loop and handles command calls.

classmethod run_process(*args, **kwargs)¶

Blockingly create and run the Serf.

This should be used as the target of a multiprocessing.Process.

static sentry_exc(exc: Exception)¶

class royalnet.serf.AlchemyConfig(database_url: str, master_table: type, identity_table: type, identity_column: str)¶: A helper class to configure Alchemy in a Serf.

Utils¶

async royalnet.utils.asyncify(function: Callable, *args, loop: Optional[asyncio.events.AbstractEventLoop] = None, **kwargs)¶: Asyncronously run the function in a different thread or process, preventing it from blocking the event loop.

Warning

If the function has side effects, it may behave strangely.

royalnet.utils.safeformat(string: str, **words: str) → str¶

str.format() something, but ignore missing keys instead of raising an error.

Parameters

string – The base string to be formatted.
words – The words to format the string with.

Returns

The formatted string.

async royalnet.utils.sleep_until(dt: datetime.datetime) → None¶: Sleep until the specified datetime.

Warning

Accurate only to seconds.

royalnet.utils.andformat(l: Collection[str], middle=', ', final=' and ') → str¶

Convert a iterable (such as a list) to a str by adding final between the last two elements and middle between the others.

Parameters

l – the input iterable.
middle – the str to be added between the middle elements.
final – the str to be added between the last two elements.

Returns

The resulting str.

Examples

>>> andformat(["Steffo", "Kappa", "Proto"])
"Steffo, Kappa and Proto"

>>> andformat(["Viktya", "Sensei", "Cate"], final=" e ")
"Viktya, Sensei e Cate"

>>> andformat(["Paltri", "Spaggia", "Gesù", "Mallllco"], middle="+", final="+")
"Paltri+Spaggia+Gesù+Mallllco"

royalnet.utils.underscorize(string: str) → str¶

Replace all non-word characters in a str with underscores.

It is particularly useful when you want to use random strings from the Internet as filenames.

Parameters: string – the input string.
Returns: The resulting string.

Example

>>> underscorize("LE EPIC PRANK [GONE WRONG!?!?]")
"LE EPIC PRANK _GONE WRONG_____"

royalnet.utils.ytdldateformat(string: Optional[str], separator: str = '-') → str¶

Convert the date str returned by youtube-dl into the YYYY-MM-DD format.

Parameters

string – the input string, in the YYYYMMDD format used by youtube_dl.
separator – the string to add between the years, the months and the days. Defaults to -.

Returns

The resulting string in the new format.

Example

>>> ytdldateformat("20111111")
"2011-11-11"

>>> ytdldateformat("20200202", separator=".")
"2020.02.02"

royalnet.utils.numberemojiformat(l: List[str]) → str¶

Convert a list to a Unicode string with one item on every line numbered with emojis.

Parameters: l – the list to convert.
Returns: The resulting Unicode string.

Examples

Cannot be displayed, as Sphinx does not render emojis properly.

royalnet.utils.ordinalformat(number: int) → str¶

Convert a int to the corresponding English ordinal str.

Parameters: number – the number to convert.
Returns: //en.wikipedia.org/wiki/Ordinal_numeral>`_.
Return type: The corresponding English `ordinal numeral <https

Examples

>>> ordinalformat(1)
"1st"
>>> ordinalformat(2)
"2nd"
>>> ordinalformat(11)
"11th"
>>> ordinalformat(101)
"101st"
>>> ordinalformat(112)
"112th"
>>> ordinalformat(0)
"0th"

royalnet.utils.to_urluuid(uuid: uuid.UUID) → str¶: Return a base64 url-friendly short UUID.

royalnet.utils.from_urluuid(b: str) → uuid.UUID¶

class royalnet.utils.MultiLock¶

A lock that can allow both simultaneous access and exclusive access to a resource.

exclusive()¶: Acquire the lock for exclusive access.

normal()¶: Acquire the lock for simultaneous access.