awebus 1.0a6

Description:

awebus 1.0a6

# awebus`awebus` (Asynchronous, Weak Event BUS) is an event system for Python whichprovides (optional) asynchronous execution of handlers, and uses weakreferences to store event handlers.## Requirements* Python >= 3.4.0## Getting Started1. Install awebus with `pip`: ```sh pip install awebus ```1. Start using awebus!## Examples### Basic Usage```pythonimport awebus# Create a busbus = awebus.Bus()# Create a handlerdef handleMyEvent(): print( "Handling My Event" ) return "Event Handled"# Register an event handlerbus.on( 'my-event', handleMyEvent )# Invoke `my-event`results = bus.emit( 'my-event' )print( results )# >> [ "Event Handled" ]# Remove the event handlerbus.off( 'my-event', handleMyEvent )```### Event ArgumentsPass arguments to event handlers as an argument list or keyword arguments.```pythonimport awebus# Create a busbus = awebus.Bus()# Create a handlerdef handleArgsEvent( foo, bar, *args, **kwargs ): print( "foo is", str( foo ) ) print( "bar is", str( bar ) ) print( "*args is", str( args ) ) print( "**kwargs is", str( kwargs ) )# Register an event handlerbus.on( 'args-event', handleArgsEvent )# Invoke `args-event` with positional, listed, and keyword argumentsbus.emit( 'args-event', 3, 7, "hello", "world", age = 30, name = "Vector" )# >> foo is 3# >> bar is 7# >> *args is ( "hello", "world" )# >> **kwargs is { "age": 30, "name": "Vector" }```### Synchronous ExecutionInvoke a synchronous event by using the `bus.emit()` method.If `emit()` encounters an asynchronous handler, it will use an asyncioevent loop to run the handler to completion synchronously.```pythonimport awebusimport asyncio# Create a busbus = awebus.Bus()# Create a regular handlerdef synchronousHandler(): print( "Synchronous Handler" ) return "sync"# Create an asynchronous handlerasync def asynchronousHandler(): await asyncio.sleep( 0.2 ) return "async"# Register handlersbus.on( 'async-event', synchronousHandler, asynchronousHandler )# Invoke the event synchronously.results = bus.emit( 'async-event' )print( results )# >> [ "sync", "async" ]```### Asynchronous ExecutionInvoke an asynchronous event by using the `bus.emitAsync()` method.If `emitAsync()` encounters a synchronous handler, it will wrap the handlerwith the asyncio coroutine decorator. `emitAsync()` returns a list ofawaitables.```pythonimport awebusimport asyncio# Create a busbus = awebus.Bus()# Create a regular handlerdef synchronousHandler(): print( "Synchronous Handler" ) return "sync"# Create an asynchronous handlerasync def asynchronousHandler(): await asyncio.sleep( 0.2 ) return "async"# Register handlersbus.on( 'async-event', synchronousHandler, asynchronousHandler )# (Inside an async method somewhere...) invoke the event asynchronously.results = await bus.emitAsync( 'async-event' )print( results )# >> [ "sync", "async" ]```### Weak ReferencesRegistered event handlers are stored internally using Python's weak referencemethod (`weakref.ref()`). This allows garbage collection of methods that wouldotherwise be cleaned up if they did not have a handler registered in the bus.If an event is invoked and a weak reference to an event handler no longerresolves, that handler is skipped and the reference to that handler isunregistered.```pythonimport awebusimport asyncio# Create a busbus = awebus.Bus()# Create a regular handlerdef handler(): print( "In Handler" ) return "handled"# Register the handlerbus.on( 'weak-event', handler )# Invoke the event# The handler will run normally at this point.results = bus.emit( 'weak-event' )print( results )# >> [ "handled" ]# Delete the handlerdel handler# Invoke the event again# The handler was never unregistered, but because weakrefs are# used, the deleted handler will be skipped and cleaned up. Yay!results = bus.emit( 'weak-event' )print( results )# >> []```#### Disabling Weak ReferencesThe use of weak references can be disabled by setting the `event_use_weakref`keyword argument to `False` when creating an event bus.```pythonimport awebusimport asyncio# Create a busbus = awebus.Bus( event_use_weakref = False )# Create a regular handlerdef handler(): print( "In Handler" ) return "result from non-weakref handler"# Register the handlerbus.on( 'no-weakref-event', handler )# Delete the reference to the handlerdel handler# Invoke the eventresults = bus.emit( 'no-weakref-event' )```### EventMixinThe module also exposes an `EventMixin` class which can be used by your ownclasses to add the event bus functionality to them with little effort.```pythonimport awebus# Create a custom classclass CustomClass( awebus.EventMixin ): def __init__( self, *args, **kwargs ): awebus.EventMixin.__init__( self, *args, **kwargs ) def onBar( self ): print( "Bar Handler" )# Create an instanceinstance = CustomClass()# Treat the instance like an event businstance.on( 'bar', instance.onBar )results = instance.emit( 'bar' )print( results )# >> [ "bar" ]```## Authors* Daniel 'Vector' Kerr <[email protected]>## LicenseSee `LICENSE`

License

For personal and professional use. You cannot resell or redistribute these repositories in their original state.

Share

• Share on Facebook
• Share on Twitter