Reference

Handling event-based streams.

event(name)

Register an event.

Note

Should only be used during Handle’s creation.

Can be used with explicit or implicit names:

Warning

The existance of name in signature does not mean an explicit argument is required.

class AutoHandle(callback=None, fallback=None, aware=False, sync=True)

Works just like Handle, except invoke() is private as _invoke().

class Handle(callback=None, fallback=None, aware=False, sync=True)

Bases: AutoHandle

Base class for those implementing the event protocol.

Parameters:

• callback (callable) – Called on dispatching existing events with (*args, **kwargs).
• fallback (callable) – Called on dispatching nonexisting events with (event, *args, **kwargs).
• sync (bool) – Whether not to create tasks from the result of callbacks.
• aware (bool) – Whether to look into the last frame’s local variables to find keys for creating a cached collections.namedtuple() for gathering dispatch values. Only use this if all events use a constant number of persistantly named values.

The idea behind this is being able to make classes that handle specific operations on signal. First, create a class with this as its subclass. Then, use the event() module decorator to register methods as handles. Sending data to them can be done via Handle.invoke() after instantiation.

Warning

All methods decorated with event() will be deleted from the class’ namespace.

After creating handles, end users can utilize these classes like so:

# ...

handle = Impl(apply = str.upper, callback = print)

while True:

    data = socket.receive()

    handle.invoke('receive', data) # will print('received', data)

This section will be updated with a more comprehensive example when available.

invoke

Call the function that’s supposed to handle the event.

Parameters:name (str) – The name of the event.

Any other parameters used will be passed to the function call.

class Track(sync=True)

Register callback functions against names.

Parameters:sync (bool) – Whether to use threading or asyncio for concurrency.

call(func)

Decorator for registering callbacks against the name. Use like event().

remove(func, name=None)

Remove the callback for the name.

@track.call
def hello():
    print('hi there')
# ...
track.remove(hello)

wait(name)

Decorator for registering temporary callbacks against the name. Use like event().

Warning

The final result is not a function; it’s an asyncio.Event or threading.Event that should be set when appropriate; only then will each callback be forgotten.

result = None

@track.wait
def on_receive(data):
    if not data.startswith('.'):
        return
    nonlocal result
    result = data
    on_receive.set()

on_receive.wait()
print('done with', result)

invoke(name, *args, **kwargs)

Call all registered functions against this name with the arguments. If sync was set to false, coroutines are gathered and scheduled as a future which is returned instead of a tuple of results.