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.

Usage:

import asab

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

Persistent dictionary

class asab.pdict.PersistentDict(path)

Bases: collections.abc.MutableMapping

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 can only store objects in the persistent dictionary that are seriazable.

load([keys]) → [values].

Optimised version of the get() operations that load multiple keys from the persistent store at once. It saves IO in exchange for possible race conditions.

Parameters:keys – A list of keys. Returns:A list of values in the same order to provided key list.

v1, v2, v3 = pdict.load('k1', 'k2', 'k3')

update([E, ]**F) → 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.6/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)

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)