Various utility classes

Singleton

class asab.abc.singleton.Singleton

The singleton pattern is a software design pattern that restricts the instantiation of a class to one object.

Note: The implementation idea is borrowed from “Creating a singleton in Python” question on StackOverflow.

classmethod delete(singleton_cls)

The method for an intentional removal of the singleton object. It shouldn’t be used unless you really know what you are doing.

One use case is a unit test, which removes an Application object after each iteration.

Usage:

import asab

class MyClass(metaclass=asab.Singleton):
    ...

Persistent dictionary

class asab.pdict.PersistentDict(path)

Bases: dict

The persistent dictionary works as the regular Python dictionary but the content of the dictionary is stored in the file. You cat think of a PersistentDict as a simple key-value store. It is not optimized for a frequent access. This class provides common dict interface.

Warning: You must explicitly load() and store() content of the dictionary Warning: You can only store objects in the persistent dictionary that are serializable.

load() → None

Load content of file as dictionary.

store() → None

Explicitly store content of persistent dictionary to file

update(other=(), **kwds) → None

Update D from mapping/iterable E and F. * If E present and has a .keys() method, does: for k in E: D[k] = E[k] * If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v * In either case, this is followed by: for k, v in F.items(): D[k] = v

Inspired by a https://github.com/python/cpython/blob/3.8/Lib/_collections_abc.py

Note: A recommended way of initializing the persistent dictionary:

PersistentState = asab.PersistentDict("some.file")
PersistentState.setdefault('foo', 0)
PersistentState.setdefault('bar', 2)

Timer

class asab.timer.Timer(app, handler, autorestart=False) → Timer.

Bases: object

The relative and optionally repeating timer for asyncio.

This class is simple relative timer that generate an event after a given time, and optionally repeating in regular intervals after that.

Parameters:

• app – An ASAB application.
• handler – A coro or future that will be called when a timer triggers.
• autorestart (boolean) – If True then a timer will be automatically restarted after triggering.

Variables:

• Handler – A coro or future that will be called when a timer triggers.
• Task – A future that represent the timer task.
• App – An ASAB app.
• AutoRestart (boolean) – If True then a timer will be automatically restarted after triggering.

The timer object is initialized as stopped.

Note: The implementation idea is borrowed from “Python - Timer with asyncio/coroutine” question on StackOverflow.

is_started() → boolean

Return True is the timer is started otherwise returns False.

restart(timeout)

Restart the timer.

Parameters:timeout (float/int) – A timer delay in seconds.

start(timeout)

Start the timer.

Parameters:timeout (float/int) – A timer delay in seconds.

stop()

Stop the timer.

Sockets

class asab.socket.StreamSocketServerService(app, service_name)

Bases: asab.abc.service.Service

Example of use:

class ServiceMyProtocolServer(asab.StreamSocketServerService):

async def initialize(self, app):

host = asab.Config.get(‘http’, ‘host’) port = asab.Config.getint(‘http’, ‘port’)

L.debug(“Starting server on {} {} …”.format(host, port)) await self.create_server(app, MyProtocol, [(host, port)])

create_server(app, protocol, addrs)

finalize(app)