bestagon 0.6.1

DDD+ES+CQRS framework for python

Bestagon

Bestagon is an asynchronous framework for quick development of microservices in a paradigm of DDD + ES + CQSR. It provides necessary building blocks for development an event sourced system:

• Event sourced Aggregate.
• Single event-sourced repository to store and retreive aggregates.
• Abstract Event Store and its implementation using KurrentDB.
• Event-sourced Application to implement use cases.
• Projection to implement views using CQRS pattern.

The framework is event-store agnostic, it contains abstract class that can be iplemented with required technology and concrete impementation with KurrentDB.

Official docs comming soon.

WARNING - the project is still in the development stage, some of functionality can change with time.

Installation

pip install bestagon

Synopsis

To create an event-sourced system you need several components:

• Aggregates
• One or multiple applications for your use cases.
• One or multiple projections to create views from your events.
• Checkpoint store
• Event store
• System to run applications and projections.
• A REST API as a main entrypoint to your system.

Use Aggregate class co define event-sourced aggregates according to your domain model:

# Define commands according to your domain model
@dataclass(frozen=True)
class IssueBond(Command):
    isin: str
    issue_date: str
    maturity_date: str
    coupon: float


@register_aggregate_type()
class Bond(Aggregate):
    
    # Aggregate always begins its lifecycle with `Created` event
    @dataclass(frozen=True)
    @register_event_type('BondIssued')
    class Issued(Aggregate.Created):
        isin: str
        issue_date: str
        maturity_date: str
        coupon: float
    
    # All consequitive events should inherit `Aggregate.Event` class
    @dataclass(frozen=True)
    @register_event_type('BondTermToMaturityChanged')
    class TermToMaturityChanged(Aggregate.Event):
        isin: str
        old_ttm: int
        new_ttm: int

    def __init__(self, event: Issued):
        super().__init__(event)
        self.isin = event.isin
        self.issue_date = date.fromisoformat(event.issue_date)
        self.maturity_date = date.fromisoformat(event.maturity_date)
        self.coupon = event.coupon
        self.ttm = None
    
    # Reactions to aggregate's own events are specified in `apply_event` method
    def apply_event(self, event: DomainEvent) -> None:
        event_routing = {
            self.TermToMaturityChanged: self._when_term_to_maturity_changed
        }
        handler = event_routing.get(type(event))
        if handler is not None:
            handler(event)
        else:
            super().apply_event(event)
    
    # Each aggregate must provide GLOBALLY UNIQUE id
    @staticmethod
    def create_id(isin: str) -> str:
        aggregate_uid = uuid5(NAMESPACE_URL, f'/bond/{isin}')
        return str(aggregate_uid)
    
    # Aggregate type is used during persistence step and should not be changed
    @staticmethod
    def get_aggregate_type() -> str:
        return 'bond'
    
    # You should not instantiate aggregate direcly, use factory method to create aggregate
    @classmethod
    def issue(cls, command: IssueBond) -> Bond:
        metadata = DomainEventMetadata(
            timestamp=DomainEventMetadata.create_timestamp(),
            aggregate_id=cls.create_id(isin=command.isin),
            aggregate_version=cls.INITIAL_VERSION,
            aggregate_type=cls.get_aggregate_type()
        )
        event = cls.Issued(
            metadata=metadata,
            isin=command.isin,
            issue_date=command.issue_date,
            maturity_date=command.maturity_date,
            coupon=command.coupon
        )
        obj = cls._create(event)
        return obj
    
    # Business logic should be executed BEFORE subsequent event creation
    def update_term_to_maturity(self) -> None:
        today = date.today()
        ttm = max([(self.maturity_date - today).days, 0])

        if ttm != self.ttm:
            metadata = DomainEventMetadata(
                timestamp=DomainEventMetadata.create_timestamp(),
                aggregate_id=self.aggregate_id,
                aggregate_version=self.next_version,
                aggregate_type=self.aggregate_type
            )
            event = self.TermToMaturityChanged(
                metadata=metadata,
                isin=str(self.isin),
                old_ttm=self.ttm,
                new_ttm=ttm
            )
            self.trigger_event(event)

Implement use cases using Application class. You can define multiple applications if necessary. To react to events generated by aggregates reimplement get_event_routing method.

class BondsApplication(Application):
    # Reaction to specific event, will be executed when bond issued
    async def _when_bond_issued(self, event: Bond.Issued) -> None:
        await self.update_term_to_maturity(isin=event.isin)
    
    # Reimplement to define events you want to react to
    def get_event_routing(self):
        event_routing = {
            Bond.Issued: self._when_bond_issued,
        }
        return event_routing
    
    # Example of a getter to retreive aggregate
    async def get_bond(self, isin: str) -> Bond:
        aggregate_id = Bond.create_id(isin=command.isin)
        bond = await self.repository.get_by_id(aggregate_type=Bond.get_aggregate_type(), aggregate_id=aggregate_id)
        return bond
    
    # For internal usages, should not be changed during lifecycle
    def get_name(self) -> str:
        return 'bonds_application'
    
    # Use case you issue a Bond
    async def issue_bond(self, command: IssueBond) -> None:
        aggregate_id = Bond.create_id(isin=command.isin)
        if await self.repository.contains(aggregate_type=Bond.get_aggregate_type(), aggregate_id=aggregate_id):
            logging.info(f'Duplicate Bond {command.isin}')
            return
        
        bond = Bond.issue(command)
        await self.repository.save(bond)
    
    # Use case to update term to maturity of a bond
    async def update_term_to_maturity(self, isin: str) -> None:
        bond = self.get_bond(isin=isin)
        bond.update_term_to_maturity()
        await self.repository.save(bond)

Use Projection class to implement necesary views for your system. You can create projection using any technology you like - MySQL, Postgres, Neo4j, etc. THe framework also provides concrete implementation for Neo4J database - Neo4JProjection.

@dataclass(frozen=True)
class GetBondsQuery(Query):
    isins: List[str]
    issue_date_start: date
    issue_date_end: date
    maturity_date_start: date
    maturity_date_end: date

    
class BondsProjection(Projection):
    async def _when_bond_issued(self, event: BondIssued) -> None:
        await self.add_bond(
            isin=event.isin, 
            issue_date=event.issue_date,
            maturity_date=event.maturity_date,
            coupon=event.coupon
        )
        
    async def _when_ttm_changed(self, event: Bond.TermToMaturityChanged) -> None:
        await self.set_term_to_maturity(isin=event.isin, ttm=event.new_ttm)
    
    async def add_bond(self, isin: str, issue_date: str, maturity_date: str, coupon: float) -> None:
        """
        Your logic of adding record to database is here
        """
        
    async def drop(self) -> None:
        """
        Your logic to drop projection is here
        """

    async def get_bonds(self, query: GetBondsQuery) -> List[dict]:
        """
        Your logic to retreive data from database is here
        """
        
    def get_event_routing(self) -> dict:
        routing = {
            Bond.Issued: self._when_bond_issued,
            Bond.TermToMaturityChanged: self._when_ttm_changed
        }
        return routing

    async def set_term_to_maturity(self, isin: str, ttm: int) -> None:
        """
        Your logic to update data in database here
        """

The next step is to define an event-sourced system:

class BondsSystem(EventSourcedSystem):
    def __init__(self, checkpoint_store, event_store):
        self.repository = EventSourcedRepository(event_store)
        
        # Instantiate your applications
        self.bonds_application = BondsApplication(repository, checkpoint_store)
        
        # Instantiate projections
        self.bonds_projection = BondsProjection(checkpint_store)
    
    # Register command handlers
    def register_command_handlers(self) -> None:
        mapper.register_command_handler(IssueBond, self.bonds_application.issue_bond)
    
    # Register query handlers
    def register_query_handler(self) -> None:
        mapper.register_query_handler(GetBondsQuery, self.bonds_projction.get_bonds)

    async def initialize(self):
        super().initialize()
        
        # You need to add your applications and projections to the system during initialization step
        await self.add_application(self.bonds_application)
        await self.add_projection(self.bonds_projection)

The last step is to define your REST API:

system = BondsSystem(
    event_store=AsyncKurrentDBEventStore(kdb_client),
    checkpoint_store=Neo4jCheckpointStore(neo4j_driver)
)

@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
    await system.initialize()
    yield
    await system.shutdown()

app = FastAPI(lifespan=lifespan)


@dataclass
class IssueBondBody:
    isin: str
    issue_date: str
    maturity_date: str
    coupon: str


@app.post('/issue_bond')
async def issue_bond(body: IssueBondBody):
    command = IssueBond(
        isin=body.isin,
        issue_date=body.issue_date,
        maturity_date=body.maturity_date,
        coupon=body.coupon
    )
    await system.execute_command(command)