The CQRS and Event Sourcing framework for Python
Project description
# Kant Framework
[![Build Status](https://travis-ci.org/patrickporto/kant.svg?branch=master)](https://travis-ci.org/patrickporto/kant)
[![codecov.io](https://codecov.io/github/patrickporto/kant/coverage.svg?branch=master)](https://codecov.io/github/patrickporto/kant?branch=master)
[![PyPI Package latest release](https://img.shields.io/pypi/v/kant.svg)](https://pypi-hypernode.com/pypi/kant)
[![Supported versions](https://img.shields.io/pypi/pyversions/kant.svg)](https://pypi-hypernode.com/pypi/kant)
[![Supported implementations](https://img.shields.io/pypi/implementation/kant.svg)](https://pypi-hypernode.com/pypi/kant)
A CQRS and Event Sourcing framework, safe for humans.
## Feature Support
* Event Store
* Optimistic concurrency control
* JSON serialization
* SQLAlchemy Projections
* Snapshots **[IN PROGRESS]**
Kant officially supports Python 3.5-3.6.
## Getting started
Create declarative events
```python
from kant import events
class BankAccountCreated(events.Event):
id = events.CUIDField(primary_key=True)
owner = events.CharField()
class DepositPerformed(events.Event):
amount = events.DecimalField()
```
Create aggregate to apply events
```python
from kant import aggregates
class BankAccount(aggregates.Aggregate):
id = aggregates.CUIDField()
owner = aggregates.CharField()
balance = aggregates.DecimalField()
def apply_bank_account_created(self, event):
self.id = event.id
self.owner = event.owner
self.balance = 0
def apply_deposit_performed(self, event):
self.balance += event.amount
```
Now, save the events
```python
from kant.eventstore import connect
await connect(user='user', password='user', database='database')
# create event store for bank_account
conn.create_keyspace('bank_account')
# create events
bank_account_created = BankAccountCreated(
id=123,
owner='John Doe',
)
deposit_performed = DepositPerformed(
amount=20,
)
bank_account = BankAccount()
bank_account.dispatch([bank_account_created, deposit_performed])
bank_account.save()
stored_bank_account = BankAccount.objects.get(123)
```
## Installing
To install Kant, simply use [pipenv](pipenv.org) (or pip)
```bash
$ pipenv install kant
```
## Contributing
Please, read the contribute guide [CONTRIBUTING](CONTRIBUTING.md).
[![Build Status](https://travis-ci.org/patrickporto/kant.svg?branch=master)](https://travis-ci.org/patrickporto/kant)
[![codecov.io](https://codecov.io/github/patrickporto/kant/coverage.svg?branch=master)](https://codecov.io/github/patrickporto/kant?branch=master)
[![PyPI Package latest release](https://img.shields.io/pypi/v/kant.svg)](https://pypi-hypernode.com/pypi/kant)
[![Supported versions](https://img.shields.io/pypi/pyversions/kant.svg)](https://pypi-hypernode.com/pypi/kant)
[![Supported implementations](https://img.shields.io/pypi/implementation/kant.svg)](https://pypi-hypernode.com/pypi/kant)
A CQRS and Event Sourcing framework, safe for humans.
## Feature Support
* Event Store
* Optimistic concurrency control
* JSON serialization
* SQLAlchemy Projections
* Snapshots **[IN PROGRESS]**
Kant officially supports Python 3.5-3.6.
## Getting started
Create declarative events
```python
from kant import events
class BankAccountCreated(events.Event):
id = events.CUIDField(primary_key=True)
owner = events.CharField()
class DepositPerformed(events.Event):
amount = events.DecimalField()
```
Create aggregate to apply events
```python
from kant import aggregates
class BankAccount(aggregates.Aggregate):
id = aggregates.CUIDField()
owner = aggregates.CharField()
balance = aggregates.DecimalField()
def apply_bank_account_created(self, event):
self.id = event.id
self.owner = event.owner
self.balance = 0
def apply_deposit_performed(self, event):
self.balance += event.amount
```
Now, save the events
```python
from kant.eventstore import connect
await connect(user='user', password='user', database='database')
# create event store for bank_account
conn.create_keyspace('bank_account')
# create events
bank_account_created = BankAccountCreated(
id=123,
owner='John Doe',
)
deposit_performed = DepositPerformed(
amount=20,
)
bank_account = BankAccount()
bank_account.dispatch([bank_account_created, deposit_performed])
bank_account.save()
stored_bank_account = BankAccount.objects.get(123)
```
## Installing
To install Kant, simply use [pipenv](pipenv.org) (or pip)
```bash
$ pipenv install kant
```
## Contributing
Please, read the contribute guide [CONTRIBUTING](CONTRIBUTING.md).
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
kant-2.1.0.tar.gz
(9.0 kB
view details)
File details
Details for the file kant-2.1.0.tar.gz
.
File metadata
- Download URL: kant-2.1.0.tar.gz
- Upload date:
- Size: 9.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 7ceac36edfeddb8ab55fefc0d8853dcf03d0c89bb594c2d069cd556e5c5d3075 |
|
MD5 | c71be7ba34b74eaa6169ddacc864d51a |
|
BLAKE2b-256 | b608c104b9c0170b676b24c21100695271fff7853c56bb18ea3792b88f373cf9 |