Metadata-Version: 2.1
Name: buslane
Version: 0.0.5
Summary: UNKNOWN
Home-page: https://github.com/konradhalas/buslane
Author: Konrad Hałas
Author-email: halas.konrad@gmail.com
License: MIT
Platform: UNKNOWN
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.6
Description-Content-Type: text/markdown

# buslane

[![Build Status](https://travis-ci.org/konradhalas/buslane.svg?branch=master)](https://travis-ci.org/konradhalas/buslane)
[![License](https://img.shields.io/pypi/l/buslane.svg)](https://pypi.python.org/pypi/buslane/)
[![Version](https://img.shields.io/pypi/v/buslane.svg)](https://pypi.python.org/pypi/buslane/)
[![Python versions](https://img.shields.io/pypi/pyversions/buslane.svg)](https://pypi.python.org/pypi/buslane/)

Simple message (event/command) bus.

## Installation

To install `buslane`, simply use `pip` (or `pipenv`):

```
$ pip install buslane
```

## Requirements

Minimum Python version supported by `buslane` is 3.6.

## Quick start

```python
from dataclasses import dataclass

from buslane.commands import Command, CommandHandler, CommandBus


@dataclass
class RegisterUserCommand(Command):
    email: str
    password: str


class RegisterUserCommandHandler(CommandHandler[RegisterUserCommand]):

    def handle(self, command: RegisterUserCommand) -> None:
        assert command == RegisterUserCommand(
            email='john@lennon.com',
            password='secret',
        )


command_bus = CommandBus()
command_bus.register(handler=RegisterUserCommandHandler())
command_bus.execute(command=RegisterUserCommand(
    email='john@lennon.com',
    password='secret',
))
```

## About

This library makes it easier to create solutions based on messages. If you want to split event occurrence from its
handling, `buslane` is the way to go. It supports commands (single handler) and events (0 or multiple handlers)
approach.

## Motivation

This package could be probably replaced with a simple Python dictionary with messages classes as keys and ordinary
functions as values. Python is a dynamic language and we can implement such solution very easy, without any classes,
inheritance, methods overriding and so one. So why you should use `buslane`? Is it the *pythonic* approach?

First of all, `buslane` is very simple and tiny project. I was copying these few lines from project to project, so now I
don't have to.

Secondly, I'm a huge fan of type annotations. This a game changer in a project with a huge codebase. `buslane` has
type hints everywhere and it is based on [Python generics][python-generics]. In combination with such tools like
[`mypy`][mypy] you will be sure that you are doing (from types point of view) everything OK.

Message handler is a class instead of function, because you can easily inject your dependencies via `__init__`
parameters. Such class is very easy to test, you don't have to `mock.patch` everything. The interface is clear.

Last but not least - the `buslane` API is simple and well defined. You can extend it easily, e.g. log all of your
messages or store them in a database.

It can be used as a foundation of your CQRS-based system.

## Reference

`buslane` uses Python type annotations to properly register handler. To create your message you have to inherit from
`Event` or `Command` class. Handler should inherit from `EventHandler[T]` or `CommandHandler[T]`, where `T` is a class
of your message.

### Events

You can register multiple or none handlers for a single event.

Classes:

- `Event`
- `EventHandler[Event]`
- `EventBus`

Exceptions:

- `WrongHandlerException`

#### Example

```python
from buslane.events import Event, EventHandler, EventBus


class SampleEvent(Event):
    pass


class SampleEventHandler(EventHandler[SampleEvent]):

    def handle(self, event: SampleEvent) -> None:
        pass


event_bus = EventBus()
event_bus.register(handler=SampleEventHandler())
event_bus.publish(event=SampleEvent())
```

### Commands

You have to register only single handler for the given command.

Classes:

- `Command`
- `CommandHandler[Command]`
- `CommandBus`

Exceptions:

- `MissingCommandHandlerException`
- `CommandHandlerRegisteredException`
- `WrongHandlerException`

#### Example

```python
from buslane.commands import Command, CommandHandler, CommandBus


class SampleCommand(Command):
    pass


class SampleCommandHandler(CommandHandler[SampleCommand]):

    def handle(self, command: SampleCommand) -> None:
        pass


command_bus = CommandBus()
command_bus.register(handler=SampleCommandHandler())
command_bus.execute(command=SampleCommand())
```

### Customizations

If you want to customize behaviour of your bus, you can override `handle` method from `EventBus` / `CommandBus` class.

The following example shows a bus which logs every event and process it in a thread.

```python
import logging
from concurrent.futures import ThreadPoolExecutor


class CustomEventBus(EventBus):

    def __init__(self, workers: int) -> None:
        super().__init__()
        self.logger = logging.getLogger()
        self.executor = ThreadPoolExecutor(max_workers=workers)

    def handle(self, event: Event, handler: EventHandler) -> None:
        self.logger.info(f'Handling event {event} by {handler}')
        self.executor.submit(handler.handle, event)
```

## Authors

Created by [Konrad Hałas][halas-homepage].

[halas-homepage]: https://konradhalas.pl
[python-generics]: https://docs.python.org/3/library/typing.html#generics
[mypy]: https://github.com/python/mypy/


