Python package for eventsourcing with Django.
Project description
Event Sourcing with Django
This package is a Django app that supports using the Python eventsourcing library with the Django ORM.
To use Django with your Python eventsourcing applications:
- install the Python package
eventsourcing_django
- add
'eventsourcing_django'
to your Django project'sINSTALLED_APPS
setting - migrate your database for this Django app
- set the environment variable
PERSISTENCE_MODULE
to'eventsourcing_django'
See below for more information.
Installation
Use pip to install the stable distribution from the Python Package Index. Please note, it is recommended to install Python packages into a Python virtual environment.
$ pip install eventsourcing_django
Alternatively, add eventsourcing_django
to your project's pyproject.yaml
or requirements.txt
file and update your virtual environment accordingly.
Event sourcing application
Define event-sourced aggregates and applications using the Application
and
Aggregate
classes from the eventsourcing
package.
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event
from uuid import uuid5, NAMESPACE_URL
class TrainingSchool(Application):
def register(self, name):
dog = Dog(name)
self.save(dog)
def add_trick(self, name, trick):
dog = self.repository.get(Dog.create_id(name))
dog.add_trick(trick)
self.save(dog)
def get_tricks(self, name):
dog = self.repository.get(Dog.create_id(name))
return dog.tricks
class Dog(Aggregate):
@event('Registered')
def __init__(self, name):
self.name = name
self.tricks = []
@staticmethod
def create_id(name):
return uuid5(NAMESPACE_URL, f'/dogs/{name}')
@event('TrickAdded')
def add_trick(self, trick):
self.tricks.append(trick)
The event sourcing application can be developed and tested independently of Django.
Next, let's configure a Django project, and our event sourcing application, so that events of the event sourcing application are stored in a Django database.
Django project settings
Add 'eventsourcing_django'
to your Django project's INSTALLED_APPS
setting.
INSTALLED_APPS = [
...
'eventsourcing_django',
]
This will make the Django models for storing events available in your Django project, and allow Django to create tables in your database for storing events.
Django database migration
Run Django's manage.py migrate
command to create database tables for storing events.
$ python manage.py migrate
Use the --database
option to create tables in a non-default database. The database
alias must be a key in the DATABASES
setting of your Django project.
$ python manage.py migrate --database=postgres
Alternatively, after the Django framework has been set up for your project, you
can call Django's call_command()
function to create the database tables.
from django.core.management import call_command
call_command('migrate')
Use the database
keyword argument to create tables in a non-default database.
call_command('migrate', database='postgres')
To set up the Django framework for your Django project, django.setup()
must have
been called after setting environment variable DJANGO_SETTINGS_MODULE
to indicate the
settings module of your Django project. This is often done by a Django project's
manage.py
, wsgi.py
, and asgi.py
files, and by tools that support Django users
such as test suite runners provided by IDEs that support Django. Django test suites
usually automatically create and migrate databases when tests are run.
Event sourcing in Django
The event sourcing application can be configured to store events in the Django project's
database by setting the environment variable PERSISTENCE_MODULE
to
'eventsourcing_django'
. This step also depends on the Django framework having been
set up to for your Django project, but it doesn't depend on the database tables having
been created.
training_school = TrainingSchool(
env={'PERSISTENCE_MODULE': 'eventsourcing_django'},
)
Use the application environment variable DJANGO_DB_ALIAS
to configure the application
to store events in a non-default Django project database. The value of DJANGO_DB_ALIAS
must correspond to one of the keys in the DATABASES
setting of the Django project.
training_school = TrainingSchool(
env={
'PERSISTENCE_MODULE': 'eventsourcing_django',
'DJANGO_DB_ALIAS': 'postgres',
}
)
You may wish to define your event sourcing application in a separate Django app,
and construct your event sourcing application in a Django AppConfig
subclass
in its apps.py
module.
# In your apps.py file.
from django.apps import AppConfig
class TrainingSchoolConfig(AppConfig):
name = '<django-project-name>.training_school'
def ready(self):
self.training_school = TrainingSchool(
env={'PERSISTENCE_MODULE': 'eventsourcing_django'}
)
You may also wish to centralize the definition of your event sourcing application's environment variables in your Django project's settings module, and use this when constructing the event sourcing application.
# Create secret cipher key.
import os
from eventsourcing.cipher import AESCipher
os.environ['CIPHER_KEY'] = AESCipher.create_key(32)
# In your settings.py file.
import os
EVENT_SOURCING_APPLICATION = {
'PERSISTENCE_MODULE': 'eventsourcing_django',
'DJANGO_DB_ALIAS': 'postgres',
'IS_SNAPSHOTTING_ENABLED': 'y',
'COMPRESSOR_TOPIC': 'eventsourcing.compressor:ZlibCompressor',
'CIPHER_TOPIC': 'eventsourcing.cipher:AESCipher',
'CIPHER_KEY': os.environ['CIPHER_KEY'],
}
# In your apps.py file.
from django.apps import AppConfig
from django.conf import settings
class TrainingSchoolConfig(AppConfig):
name = '<django-project-name>.training_school'
def ready(self):
self.training_school = TrainingSchool(env=settings.EVENT_SOURCING_APPLICATION)
The single instance of the event sourcing application can then be obtained in other places, such as views, forms, management commands, and tests.
from django.apps import apps
training_school = apps.get_app_config('training_school').training_school
The event sourcing application's methods can be called in views, forms, management commands, and tests.
training_school.register('Fido')
training_school.add_trick('Fido', 'roll over')
training_school.add_trick('Fido', 'play dead')
tricks = training_school.get_tricks('Fido')
assert tricks == ['roll over', 'play dead']
Events will be stored in the Django project's database, so long as the
database tables have been created before the event sourcing application
methods are called. If the database tables have not been created, an
OperationalError
will be raised to indicate that the tables are not found.
Summary
In summary, before constructing an event sourcing application with eventsourcing_django
as its persistence module, the Django framework must have been set up for a Django
project that has 'eventsourcing_django'
included in its INSTALLED_APPS
setting.
And, before calling the methods of the event sourcing application, the Django project's
database must have been migrated.
For more information, please refer to the Python eventsourcing library and the Django project.
Management commands
The eventsourcing_django
package is a Django app which ships with the following
Django management commands. They are available in Django projects that have
'eventsourcing_django'
included in their INSTALLED_APPS
setting.
At the moment, there is only one management command: sync_followers
.
The sync_followers
management command helps users of the eventsourcing.system
module. Please refer to the eventsourcing
package docs for more information
about the eventsourcing.system
module.
Synchronise followers
Manually synchronise followers (i.e. ProcessApplication
instances) with all of their
leaders, as defined in the eventsourcing.system.System
's pipes.
Usage
$ python manage.py sync_followers [-n] [-v {0,1,2,3}] [follower [follower ...]]
Where follower
denotes the name of a follower to synchronize. Not specifying any means
synchronising all followers found in the system.
Relevant options:
-n
,--dry-run
: Load and process all unseen events for the selected followers, but roll back all changes at the end.-v {0,1,2,3}
,--verbosity {0,1,2,3}
: Verbosity level; 0=minimal output, 1=normal output, 2=verbose output, 3=very verbose output.
For a full list of options, pass the --help
flag to the command.
Examples
-
To synchronise all followers found in the runner:
$ python manage.py sync_followers
-
To synchronise a single follower:
$ python manage.py sync_followers TrainingSchool
The command supports the regular -v/--verbosity
optional argument, as well as a
-n/--dry-run
flag.
Note that running the command in dry-run mode will pull and process every new event, though the changes will eventually be rolled back.
Error handling
Each selected follower should have its own chance at synchronisation. Therefore, the command will catch some exceptions on a per-follower basis and continue with the remaining followers.
The base Django exceptions that are caught are EmptyResultSet
, FieldDoesNotExist
,
FieldError
, MultipleObjectsReturned
, and ObjectDoesNotExist
. The base exception
EventSourcingError
from the eventsourcing
library is also caught per follower.
Configuration
This command needs to access a eventsourcing.system.Runner
instance to query and act
on its followers. The runner's system is additionally the one defining the pipes between
leaders and followers.
The default behaviour, without additional configuration, is to inspect all installed
Django apps and look for an instance of eventsourcing.system.Runner
. The attribute
name does not matter as long as it is public (i.e. not start with an underscore).
# djangoproject/apps/my_es_app/apps.py
import eventsourcing.system
from django.apps import AppConfig
class MyEventSourcedAppConfig(AppConfig):
name = 'my_event_sourced_app'
runner: eventsourcing.system.Runner
def ready(self) -> None:
self.runner = eventsourcing.system.SingleThreadedRunner(
eventsourcing.system.System(...)
)
This is usually enough unless you i) have multiple runners defined in one or more apps,
or ii) do not hold the runner(s) in Django apps. In which case, you should configure the
Django setting EVENTSOURCING_RUNNER
in one of two ways:
-
Set
EVENTSOURCING_RUNNER
to an app name's attribute. This attribute must be aeventsourcing.system.Runner
instance.# djangoproject/settings.py ... EVENTSOURCING_RUNNER = 'my_event_sourced_app.runner'
-
Set
EVENTSOURCING_RUNNER
to a fully qualified function name. This function will be called without arguments and should return aeventsourcing.system.Runner
instance.# djangoproject/settings.py ... EVENTSOURCING_RUNNER = 'djangoproject.runner_utils.get_runner'
# djangoproject/runner_utils.py import eventsourcing.system def get_runner() -> eventsourcing.system.Runner: return ...
All runner classes shipped with the eventsourcing
library are compatible.
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
Built Distribution
File details
Details for the file eventsourcing_django-0.5.1.tar.gz
.
File metadata
- Download URL: eventsourcing_django-0.5.1.tar.gz
- Upload date:
- Size: 16.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.3 CPython/3.13.0 Darwin/24.0.0
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 5e42c695b1cb60b4961de41ae2ca70d2e364a5a343de09f727b764f5df5f7141 |
|
MD5 | 1e2dc833c1036d876b96da4b80ba03b6 |
|
BLAKE2b-256 | ea2b621dc26853ff4d4cfb07e611a9a7df086051e1cdf49ee7a6f454b797e1ed |
Provenance
File details
Details for the file eventsourcing_django-0.5.1-py3-none-any.whl
.
File metadata
- Download URL: eventsourcing_django-0.5.1-py3-none-any.whl
- Upload date:
- Size: 15.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.3 CPython/3.13.0 Darwin/24.0.0
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | bafc229a31b4a542b93105250fefce5f04641651b02cf8c0435485a6a6851f75 |
|
MD5 | 7ab8443e5e481ba1c3b6fa22763cbcb7 |
|
BLAKE2b-256 | 4004e3a888b23c154dc8c8986615930fd5ecb89ddb54b2c1e7c0b7f65b0d5892 |