This package supports using the Python eventsourcing library with Amazon DynamoDB.
To use DynamoDB with your Python eventsourcing applications:
- install the Python package
eventsourcing_dynamodb
- set the environment variable
PERSISTENCE_MODULE
to'eventsourcing_dynamodb'
- set the environment variable
DYNAMO_ENDPOINT_URL
to a DynamoDB endpoint URL - set the environment variable
DYNAMO_TABLE
to a DynamoDB table
See below for more information.
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_dynamodb
Define aggregates and applications in the usual way.
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)
To use this module as the persistence module for your application, set the environment
variable PERSISTENCE_MODULE
to 'eventsourcing_dynamodb'
.
When using this module, you need to set the environment variables DYNAMO_ENDPOINT_URL
and
DYNAMO_TABLE
to a DynamoDB URL and table.
import os
os.environ['PERSISTENCE_MODULE'] = 'eventsourcing_dynamodb'
os.environ['DYNAMO_ENDPOINT_URL'] = 'http://localhost:8000'
os.environ['DYNAMO_TABLE'] = 'dynamo_events'
Construct and use the application in the usual way.
school = TrainingSchool()
school.register('Fido')
school.add_trick('Fido', 'roll over')
school.add_trick('Fido', 'play dead')
tricks = school.get_tricks('Fido')
assert tricks == ['roll over', 'play dead']