[](https://github.com/pyeventsourcing/eventsourcing/tree/main)
[](https://coveralls.io/github/pyeventsourcing/eventsourcing?branch=main)
[](https://eventsourcing.readthedocs.io/en/stable/)
[](https://pypi.org/project/eventsourcing/)
[](https://pypistats.org/packages/eventsourcing)
[](https://github.com/psf/black)
# Event Sourcing in Python
A library for event sourcing in Python.
*"totally amazing and a pleasure to use"*
*"very clean and intuitive"*
*"a huge help and time saver"*
Please [read the docs](https://eventsourcing.readthedocs.io/). See also [extension projects](https://github.com/pyeventsourcing).
## Installation
Use pip to install the [stable distribution](https://pypi.org/project/eventsourcing/)
from the Python Package Index.
$ pip install eventsourcing
Please note, it is recommended to install Python
packages into a Python virtual environment.
## Synopsis
Define aggregates with the `Aggregate` class and the `@event` decorator.
```python
from eventsourcing.domain import Aggregate, event
class Dog(Aggregate):
@event('Registered')
def __init__(self, name):
self.name = name
self.tricks = []
@event('TrickAdded')
def add_trick(self, trick):
self.tricks.append(trick)
```
Define application objects with the `Application` class.
```python
from eventsourcing.application import Application
class DogSchool(Application):
def register_dog(self, name):
dog = Dog(name)
self.save(dog)
return dog.id
def add_trick(self, dog_id, trick):
dog = self.repository.get(dog_id)
dog.add_trick(trick)
self.save(dog)
def get_dog(self, dog_id):
dog = self.repository.get(dog_id)
return {'name': dog.name, 'tricks': tuple(dog.tricks)}
```
Write a test.
```python
def test_dog_school():
# Construct application object.
school = DogSchool()
# Evolve application state.
dog_id = school.register_dog('Fido')
school.add_trick(dog_id, 'roll over')
school.add_trick(dog_id, 'play dead')
# Query application state.
dog = school.get_dog(dog_id)
assert dog['name'] == 'Fido'
assert dog['tricks'] == ('roll over', 'play dead')
# Select notifications.
notifications = school.notification_log.select(start=1, limit=10)
assert len(notifications) == 3
```
Run the test with the default persistence module. Events are stored
in memory using Python objects.
```python
test_dog_school()
```
Configure the application to run with an SQLite database. Other persistence modules are available.
```python
import os
os.environ["PERSISTENCE_MODULE"] = 'eventsourcing.sqlite'
os.environ["SQLITE_DBNAME"] = 'dog-school.db'
```
Run the test with SQLite.
```python
test_dog_school()
```
See the [documentation](https://eventsourcing.readthedocs.io/) for more information.
## Features
**Aggregates and applications** — base classes for event-sourced aggregates
and applications. Suggests how to structure an event-sourced application. All
classes are fully type-hinted to guide developers in using the library.
**Flexible event store** — flexible persistence of aggregate events. Combines
an event mapper and an event recorder in ways that can be easily extended.
Mapper uses a transcoder that can be easily extended to support custom
model object types. Recorders supporting different databases can be easily
substituted and configured with environment variables.
**Application-level encryption and compression** — encrypts and decrypts events inside the
application. This means data will be encrypted in transit across a network ("on the wire")
and at disk level including backups ("at rest"), which is a legal requirement in some
jurisdictions when dealing with personally identifiable information (PII) for example
the EU's GDPR. Compression reduces the size of stored aggregate events and snapshots, usually
by around 25% to 50% of the original size. Compression reduces the size of data
in the database and decreases transit time across a network.
**Snapshotting** — reduces access-time for aggregates that have many events.
**Versioning** - allows changes to be introduced after an application
has been deployed. Both aggregate events and aggregate snapshots can be versioned.
**Optimistic concurrency control** — ensures a distributed or horizontally scaled
application doesn't become inconsistent due to concurrent method execution. Leverages
optimistic concurrency controls in adapted database management systems.
**Notifications and projections** — reliable propagation of application
events with pull-based notifications allows the application state to be
projected accurately into replicas, indexes, view models, and other applications.
Supports materialized views and CQRS.
**Event-driven systems** — reliable event processing. Event-driven systems
can be defined independently of particular persistence infrastructure and mode of
running.
**Detailed documentation** — documentation provides general overview, introduction
of concepts, explanation of usage, and detailed descriptions of library classes.
All code is annotated with type hints.
**Worked examples** — includes examples showing how to develop aggregates, applications
and systems.
## Extensions
The GitHub organisation
[Event Sourcing in Python](https://github.com/pyeventsourcing)
hosts extension projects for the Python eventsourcing library.
There are projects that adapt popular ORMs such as
[Django](https://github.com/pyeventsourcing/eventsourcing-django#readme)
and [SQLAlchemy](https://github.com/pyeventsourcing/eventsourcing-sqlalchemy#readme).
There are projects that adapt specialist event stores such as
[Axon Server](https://github.com/pyeventsourcing/eventsourcing-axonserver#readme) and
[EventStoreDB](https://github.com/pyeventsourcing/eventsourcing-eventstoredb#readme).
There are projects that support popular NoSQL databases such as
[DynamoDB](https://github.com/pyeventsourcing/eventsourcing-dynamodb#readme).
There are also projects that provide examples of using the
library with web frameworks such as
[FastAPI](https://github.com/pyeventsourcing/example-fastapi#readme)
and [Flask](https://github.com/pyeventsourcing/example-flask#readme),
and for serving applications and running systems with efficient
inter-process communication technologies like [gRPC](https://github.com/pyeventsourcing/eventsourcing-grpc#readme).
And there are examples of event-sourced applications and systems
of event-sourced applications, such as the
[Paxos system](https://github.com/pyeventsourcing/example-paxos#readme),
which is used as the basis for a
[replicated state machine](https://github.com/pyeventsourcing/example-paxos/tree/master/replicatedstatemachine),
which is used as the basis for a
[distributed key-value store](https://github.com/pyeventsourcing/example-paxos/tree/master/keyvaluestore).
## Project
This project is [hosted on GitHub](https://github.com/pyeventsourcing/eventsourcing).
Please register questions, requests and
[issues on GitHub](https://github.com/pyeventsourcing/eventsourcing/issues),
or post in the project's Slack channel.
There is a [Slack channel](https://join.slack.com/t/eventsourcinginpython/shared_invite/zt-3hogb36o-LCvKd4Rz8JMALoLSl_pQ8g)
for this project, which you are [welcome to join](https://join.slack.com/t/eventsourcinginpython/shared_invite/zt-3hogb36o-LCvKd4Rz8JMALoLSl_pQ8g).
Please refer to the [documentation](https://eventsourcing.readthedocs.io/) for installation and usage guides.
Raw data
{
"_id": null,
"home_page": "https://github.com/pyeventsourcing/eventsourcing",
"name": "eventsourcing",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "event sourcing, event store, domain driven design, domain-driven design, ddd, cqrs, cqs",
"author": "John Bywater",
"author_email": "john.bywater@appropriatesoftware.net",
"download_url": "https://files.pythonhosted.org/packages/41/4f/ea439503e7c68edd703b1d22667f0c2cfa4c16b2d59e54d590025b38b863/eventsourcing-9.3.5.tar.gz",
"platform": null,
"description": "[](https://github.com/pyeventsourcing/eventsourcing/tree/main)\n[](https://coveralls.io/github/pyeventsourcing/eventsourcing?branch=main)\n[](https://eventsourcing.readthedocs.io/en/stable/)\n[](https://pypi.org/project/eventsourcing/)\n[](https://pypistats.org/packages/eventsourcing)\n[](https://github.com/psf/black)\n\n\n# Event Sourcing in Python\n\nA library for event sourcing in Python.\n\n*\"totally amazing and a pleasure to use\"*\n\n*\"very clean and intuitive\"*\n\n*\"a huge help and time saver\"*\n\nPlease [read the docs](https://eventsourcing.readthedocs.io/). See also [extension projects](https://github.com/pyeventsourcing).\n\n\n## Installation\n\nUse pip to install the [stable distribution](https://pypi.org/project/eventsourcing/)\nfrom the Python Package Index.\n\n $ pip install eventsourcing\n\nPlease note, it is recommended to install Python\npackages into a Python virtual environment.\n\n\n## Synopsis\n\nDefine aggregates with the `Aggregate` class and the `@event` decorator.\n\n```python\nfrom eventsourcing.domain import Aggregate, event\n\nclass Dog(Aggregate):\n @event('Registered')\n def __init__(self, name):\n self.name = name\n self.tricks = []\n\n @event('TrickAdded')\n def add_trick(self, trick):\n self.tricks.append(trick)\n```\n\nDefine application objects with the `Application` class.\n\n```python\nfrom eventsourcing.application import Application\n\nclass DogSchool(Application):\n def register_dog(self, name):\n dog = Dog(name)\n self.save(dog)\n return dog.id\n\n def add_trick(self, dog_id, trick):\n dog = self.repository.get(dog_id)\n dog.add_trick(trick)\n self.save(dog)\n\n def get_dog(self, dog_id):\n dog = self.repository.get(dog_id)\n return {'name': dog.name, 'tricks': tuple(dog.tricks)}\n```\n\nWrite a test.\n\n```python\ndef test_dog_school():\n # Construct application object.\n school = DogSchool()\n\n # Evolve application state.\n dog_id = school.register_dog('Fido')\n school.add_trick(dog_id, 'roll over')\n school.add_trick(dog_id, 'play dead')\n\n # Query application state.\n dog = school.get_dog(dog_id)\n assert dog['name'] == 'Fido'\n assert dog['tricks'] == ('roll over', 'play dead')\n\n # Select notifications.\n notifications = school.notification_log.select(start=1, limit=10)\n assert len(notifications) == 3\n```\n\nRun the test with the default persistence module. Events are stored\nin memory using Python objects.\n\n```python\ntest_dog_school()\n```\n\nConfigure the application to run with an SQLite database. Other persistence modules are available.\n\n```python\nimport os\n\nos.environ[\"PERSISTENCE_MODULE\"] = 'eventsourcing.sqlite'\nos.environ[\"SQLITE_DBNAME\"] = 'dog-school.db'\n```\n\nRun the test with SQLite.\n\n```python\ntest_dog_school()\n```\n\nSee the [documentation](https://eventsourcing.readthedocs.io/) for more information.\n\n\n## Features\n\n**Aggregates and applications** \u2014 base classes for event-sourced aggregates\nand applications. Suggests how to structure an event-sourced application. All\nclasses are fully type-hinted to guide developers in using the library.\n\n**Flexible event store** \u2014 flexible persistence of aggregate events. Combines\nan event mapper and an event recorder in ways that can be easily extended.\nMapper uses a transcoder that can be easily extended to support custom\nmodel object types. Recorders supporting different databases can be easily\nsubstituted and configured with environment variables.\n\n**Application-level encryption and compression** \u2014 encrypts and decrypts events inside the\napplication. This means data will be encrypted in transit across a network (\"on the wire\")\nand at disk level including backups (\"at rest\"), which is a legal requirement in some\njurisdictions when dealing with personally identifiable information (PII) for example\nthe EU's GDPR. Compression reduces the size of stored aggregate events and snapshots, usually\nby around 25% to 50% of the original size. Compression reduces the size of data\nin the database and decreases transit time across a network.\n\n**Snapshotting** \u2014 reduces access-time for aggregates that have many events.\n\n**Versioning** - allows changes to be introduced after an application\nhas been deployed. Both aggregate events and aggregate snapshots can be versioned.\n\n**Optimistic concurrency control** \u2014 ensures a distributed or horizontally scaled\napplication doesn't become inconsistent due to concurrent method execution. Leverages\noptimistic concurrency controls in adapted database management systems.\n\n**Notifications and projections** \u2014 reliable propagation of application\nevents with pull-based notifications allows the application state to be\nprojected accurately into replicas, indexes, view models, and other applications.\nSupports materialized views and CQRS.\n\n**Event-driven systems** \u2014 reliable event processing. Event-driven systems\ncan be defined independently of particular persistence infrastructure and mode of\nrunning.\n\n**Detailed documentation** \u2014 documentation provides general overview, introduction\nof concepts, explanation of usage, and detailed descriptions of library classes.\nAll code is annotated with type hints.\n\n**Worked examples** \u2014 includes examples showing how to develop aggregates, applications\nand systems.\n\n\n## Extensions\n\nThe GitHub organisation\n[Event Sourcing in Python](https://github.com/pyeventsourcing)\nhosts extension projects for the Python eventsourcing library.\nThere are projects that adapt popular ORMs such as\n[Django](https://github.com/pyeventsourcing/eventsourcing-django#readme)\nand [SQLAlchemy](https://github.com/pyeventsourcing/eventsourcing-sqlalchemy#readme).\nThere are projects that adapt specialist event stores such as\n[Axon Server](https://github.com/pyeventsourcing/eventsourcing-axonserver#readme) and\n[EventStoreDB](https://github.com/pyeventsourcing/eventsourcing-eventstoredb#readme).\nThere are projects that support popular NoSQL databases such as\n[DynamoDB](https://github.com/pyeventsourcing/eventsourcing-dynamodb#readme).\nThere are also projects that provide examples of using the\nlibrary with web frameworks such as\n[FastAPI](https://github.com/pyeventsourcing/example-fastapi#readme)\nand [Flask](https://github.com/pyeventsourcing/example-flask#readme),\nand for serving applications and running systems with efficient\ninter-process communication technologies like [gRPC](https://github.com/pyeventsourcing/eventsourcing-grpc#readme).\nAnd there are examples of event-sourced applications and systems\nof event-sourced applications, such as the\n[Paxos system](https://github.com/pyeventsourcing/example-paxos#readme),\nwhich is used as the basis for a\n[replicated state machine](https://github.com/pyeventsourcing/example-paxos/tree/master/replicatedstatemachine),\nwhich is used as the basis for a\n[distributed key-value store](https://github.com/pyeventsourcing/example-paxos/tree/master/keyvaluestore).\n\n## Project\n\nThis project is [hosted on GitHub](https://github.com/pyeventsourcing/eventsourcing).\n\nPlease register questions, requests and\n[issues on GitHub](https://github.com/pyeventsourcing/eventsourcing/issues),\nor post in the project's Slack channel.\n\nThere is a [Slack channel](https://join.slack.com/t/eventsourcinginpython/shared_invite/zt-3hogb36o-LCvKd4Rz8JMALoLSl_pQ8g)\nfor this project, which you are [welcome to join](https://join.slack.com/t/eventsourcinginpython/shared_invite/zt-3hogb36o-LCvKd4Rz8JMALoLSl_pQ8g).\n\nPlease refer to the [documentation](https://eventsourcing.readthedocs.io/) for installation and usage guides.\n\n",
"bugtrack_url": null,
"license": "BSD 3-Clause",
"summary": "Event sourcing in Python",
"version": "9.3.5",
"project_urls": {
"Homepage": "https://github.com/pyeventsourcing/eventsourcing",
"Repository": "https://github.com/pyeventsourcing/eventsourcing"
},
"split_keywords": [
"event sourcing",
" event store",
" domain driven design",
" domain-driven design",
" ddd",
" cqrs",
" cqs"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d16476830243518339da01d631acad42b800b0999925905ceb888af747dd458f",
"md5": "182c073113929687cefb65fc5607e937",
"sha256": "7debbf1fbf5381cd3b072356a00fd21d6bd29ccf526412e3ca061cfe3068a92b"
},
"downloads": -1,
"filename": "eventsourcing-9.3.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "182c073113929687cefb65fc5607e937",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 73970,
"upload_time": "2024-11-08T23:00:26",
"upload_time_iso_8601": "2024-11-08T23:00:26.980061Z",
"url": "https://files.pythonhosted.org/packages/d1/64/76830243518339da01d631acad42b800b0999925905ceb888af747dd458f/eventsourcing-9.3.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "414fea439503e7c68edd703b1d22667f0c2cfa4c16b2d59e54d590025b38b863",
"md5": "2c3edac2635a35404df23f7117f44bab",
"sha256": "30ba7cc28c409abff0005967e9053f2f81334acb796793806444ed081f95a4f6"
},
"downloads": -1,
"filename": "eventsourcing-9.3.5.tar.gz",
"has_sig": false,
"md5_digest": "2c3edac2635a35404df23f7117f44bab",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 69955,
"upload_time": "2024-11-08T23:00:28",
"upload_time_iso_8601": "2024-11-08T23:00:28.955852Z",
"url": "https://files.pythonhosted.org/packages/41/4f/ea439503e7c68edd703b1d22667f0c2cfa4c16b2d59e54d590025b38b863/eventsourcing-9.3.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-08 23:00:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "pyeventsourcing",
"github_project": "eventsourcing",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"lcname": "eventsourcing"
}
JSON
