API Reference¶

This page is autogenerated from the docstrings inside the code.

Alchemy¶

The subpackage providing all functions and classes related to databases and tables.

It requires either the alchemy_easy or the alchemy_hard extras to be installed.

You can install alchemy_easy with:

pip install royalnet[alchemy_easy]

To install alchemy_hard, refer to the `psycopg2 <https://pypi.org/project/psycopg2/>}`_ installation instructions, then run:

pip install royalnet[alchemy_hard]

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.ext.declarative.api.DeclarativeMeta¶

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() → AsyncIterator[sqlalchemy.orm.session.Session]¶

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() → Iterator[sqlalchemy.orm.session.Session]¶

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.TableNotFoundError¶: The requested table was not found.

Backpack¶

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

Bard¶

The subpackage providing all classes related to music files.

It requires the bard extra to be installed (the ffmpeg_python, youtube_dl and eyed3 packages).

You can install it with:

pip install royalnet[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 YoutubeDL.extract_info() function.

Warning

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

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

Fetch the info for an url through YoutubeDL.

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': './downloads/%(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.

async 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 YoutubeDL.

set_ytdlinfo_from_id3_tags()¶

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¶

The subpackage providing all classes related to Royalnet commands.

class royalnet.commands.Command(serf: Serf, config: ConfigDict)¶

property alchemy¶: A shortcut for interface.alchemy.

aliases: List[str] = []¶

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: str = NotImplemented¶: A small description of the command, to be displayed when the command is being autocompleted.

property loop¶: A shortcut for interface.loop.

name: str = NotImplemented¶

The main name of the command.

Example

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

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

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

class royalnet.commands.CommandData(command)¶

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 find_user(alias: str) → Optional[royalnet.backpack.tables.users.User]¶

Find the User having a specific Alias.

Parameters: alias – the Alias to search for.

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.

keyboard(text, keys: List[KeyboardKey])¶

property loop¶

classmethod register_keyboard_key(identifier: str, key: KeyboardKey)¶

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.

async reply_image(image: io.IOBase, caption: Optional[str] = None) → None¶

Send an image (with optionally a caption) to the channel where the call was made.

Parameters

image – The bytes of the image to send.
caption – The caption to attach to the image.

property session¶

async session_close()¶: Asyncronously close the session of this object.

async session_commit()¶: Asyncronously commit the session of this object.

classmethod unregister_keyboard_key(identifier: str)¶

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: 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.
Raises: 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.

exception royalnet.commands.ProgramError(message='')¶: The command encountered an error in the program.

class royalnet.commands.HeraldEvent(parent: Union[Serf, Constellation], config)¶

A remote procedure call triggered by a royalnet.herald request.

property alchemy¶: A shortcut for parent.alchemy.

property loop¶: A shortcut for parent.loop.

name = NotImplemented¶: The event_name that will trigger this event.

async run(**kwargs)¶

class royalnet.commands.KeyboardKey(short: str, text: str, callback: Callable[[royalnet.commands.commanddata.CommandData], Awaitable[None]])¶

async press(data: royalnet.commands.commanddata.CommandData)¶

class royalnet.commands.ConfigDict¶

classmethod convert(item)¶

Constellation¶

The subpackage providing all functions and classes that handle the webserver and the webpages.

It requires the constellation extra to be installed (starlette).

You can install it with:

pip install royalnet[constellation]

It optionally uses the sentry extra for error reporting.

You can install them with:

pip install royalnet[constellation,sentry]

class royalnet.constellation.Constellation(alchemy_cfg: Dict[str, Any], herald_cfg: Dict[str, Any], packs_cfg: Dict[str, Any], constellation_cfg: Dict[str, Any], logging_cfg: Dict[str, Any])¶

The class that represents the webserver.

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

It also handles the Alchemy connection, and Herald connections too.

address: str = None¶: The address that the Constellation will bind to when run.

alchemy = None¶: The Alchemy of this Constellation.

async call_herald_event(destination: str, event_name: str, **kwargs) → Dict¶: Send a royalherald.Request to a specific destination, and wait for a royalherald.Response.

events: Dict[str, rc.HeraldEvent] = None¶: A dictionary containing all Event that can be handled by this Constellation.

herald: Optional[rh.Link] = None¶: The Link object connecting the Constellation to the rest of the herald network. As is the case with the logging module, it will be started on the first request received by the Constellation, as the event loop won’t be available before that.

herald_task: Optional[aio.Task] = None¶: A reference to the aio.Task that runs the rh.Link.

init_herald(herald_cfg: Dict[str, Any])¶: Create a rh.Link.

loop: Optional[aio.AbstractEventLoop] = None¶

The event loop of the Constellation.

Because of how uvicorn runs, it will stay None until the first page is requested.

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

port: int = None¶: The port on which the Constellation will listen for connection on.

register_events(events: List[Type[royalnet.commands.heraldevent.HeraldEvent]], pack_cfg: Dict[str, Any])¶

register_page_stars(page_stars: List[Type[royalnet.constellation.pagestar.PageStar]], pack_cfg: Dict[str, Any])¶

run_blocking()¶

classmethod run_process(alchemy_cfg: Dict[str, Any], herald_cfg: Dict[str, Any], sentry_cfg: Dict[str, Any], packs_cfg: Dict[str, Any], constellation_cfg: Dict[str, Any], logging_cfg: Dict[str, Any])¶

Blockingly create and run the Constellation.

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

running: bool = None¶: Is the Constellation server currently running?

starlette = None¶: The Starlette app.

stars: List[PageStar] = None¶: A list of all the PageStar registered to this Constellation.

class royalnet.constellation.Star(constellation: Constellation, config: ConfigDict)¶

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.

abstract async page(request: starlette.requests.Request) → starlette.responses.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 alchemy.session_acm() of the Constellation.

class royalnet.constellation.PageStar(constellation: Constellation, config: ConfigDict)¶

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(), path and optionally methods.

classmethod methods()¶

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: str = NotImplemented¶

The route of the star.

Example

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

Herald¶

The subpackage providing all functions and classes to handle communication between process (even over the Internet).

It is based on websockets.

It requires the herald extra to be installed.

You can install it with:

pip install royalnet[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.

classmethod from_config(*, name: str, address: str, port: int, secret: str, secure: bool = False, path: str = '/', enabled: ... = Ellipsis)¶

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 config.url.

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(logging_cfg: Dict[str, Any])¶

serve()¶

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

classmethod from_dict(d: dict)¶

to_dict()¶

Serf¶

The subpackage providing all Serf implementations.

class royalnet.serf.Serf(loop: asyncio.events.AbstractEventLoop, alchemy_cfg: royalnet.commands.configdict.ConfigDict, herald_cfg: royalnet.commands.configdict.ConfigDict, packs_cfg: royalnet.commands.configdict.ConfigDict, **_)¶

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

alchemy: Optional[ra.Alchemy] = None¶: The Alchemy object connecting this Serf to a database.

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

async call_herald_event(destination: str, event_name: str, **kwargs) → Dict¶: Send a royalherald.Request to a specific destination, and wait for a royalherald.Response.

commands: Dict[str, rc.Command] = None¶: The dict connecting each command name to its Command object.

events: Dict[str, rc.HeraldEvent] = None¶: A dictionary containing all Event that can be handled by this Serf.

herald: Optional[rh.Link] = None¶: The Link object connecting the Serf to the rest of the Herald network.

herald_task: Optional[aio.Task] = None¶: A reference to the asyncio.Task that runs the Link.

property identity_chain¶: Find a relationship path starting from the master table and ending at the identity table, and return it.

identity_table: Optional[Table] = None¶: The identity table containing the interface data (such as the Telegram user data) and that is in a many-to-one relationship with the master table.

init_alchemy(alchemy_cfg: Dict[str, Any], 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_herald(herald_cfg: royalnet.commands.configdict.ConfigDict)¶: Create a Link and bind Event.

interface_name = NotImplemented¶

loop: Optional[aio.AbstractEventLoop] = None¶: The event loop this Serf is running on.

master_table: Optional[Table] = None¶: The central table listing all users. It usually is User.

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

prefix = NotImplemented¶

async static press(key: royalnet.commands.keyboardkey.KeyboardKey, data: royalnet.commands.commanddata.CommandData)¶

register_commands(commands: List[Type[royalnet.commands.command.Command]], pack_cfg: royalnet.commands.configdict.ConfigDict) → None¶: Initialize and register all commands passed as argument.

register_events(events: List[Type[royalnet.commands.heraldevent.HeraldEvent]], pack_cfg: royalnet.commands.configdict.ConfigDict)¶

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

classmethod run_process(**kwargs)¶

Blockingly create and run the Serf.

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

exception royalnet.serf.SerfError¶: Base class for all royalnet.serf errors.

Utils¶

async royalnet.utils.asyncify(function: Callable, *args, loop: Optional[asyncio.events.AbstractEventLoop] = None, **kwargs)¶

Asyncronously run the function in a executor, allowing it to run asyncronously.

Parameters

function – The function to call.
args – The arguments to pass to the function.
kwargs – The keyword arguments to pass to the function.
loop – The loop to run the function in. If None, run it in in the current event loop.

Warning

The called function must be thread-safe!

Warning

Calling a function this way might be significantly slower than calling its blocking counterpart!

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

Warning

Accurate only to seconds.

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

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

Parameters

coll – the input collection.
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 str, in the YYYYMMDD format used by youtube_dl.
separator – the str to add between the years, the months and the days. Defaults to "-".

Returns

The resulting str in the new format.

Example

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

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

royalnet.utils.numberemojiformat(li: Collection[str]) → str¶

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

Parameters: li – 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.

royalnet.utils.init_sentry(sentry_cfg: Dict[str, Any])¶

royalnet.utils.sentry_exc(exc: Exception, level: str = 'ERROR')¶

royalnet.utils.sentry_wrap(level: str = 'ERROR')¶

royalnet.utils.sentry_async_wrap(level: str = 'ERROR')¶

royalnet.utils.init_logging(logging_cfg: Dict[str, Any])¶

royalnet.utils.strip_tabs(s: str) → str¶