eventcore

A small framework for modelling Commands and Events formally in a Django application.

It sits at a deliberate midpoint: richer than scattering business logic across views and save() overrides, but not event sourcing. Models stay the source of truth and are mutated in place — events are a shared vocabulary for state transitions and the seam where side effects (Celery tasks, notifications, cascading commands) attach.

lib/ is on the Python path, so import it as a top-level package:

import eventcore
from eventcore import Command, CommandError, Event, ModelEvent, apply_event, invoke, on

Synopsis

How the pieces fit together in application code, end to end:

# orders/models.py — a plain Django model
from django.db import models


class Order(models.Model):
    ...


# orders/events.py — facts that can happen to an Order
from eventcore import ModelEvent
from orders.models import Order

OrderEvent = ModelEvent[Order]


class ShipmentCreated(OrderEvent):
    carrier_name: str

    def apply(self, model: Order) -> None:
        model.shipments.create(carrier_name=self.carrier_name)


# orders/commands.py — validated intentions that orchestrate the change
from eventcore import Command, CommandError, apply_event
from orders.events import ShipmentCreated
from orders.models import Order


class ShipOrder(Command):
    order_id: int
    carrier_name: str

    class OrderNotFound(CommandError): ...

    def perform(self):
        try:
            order = Order.objects.get(pk=self.order_id)
        except Order.DoesNotExist:
            raise self.OrderNotFound("Order %s not found", self.order_id)

        apply_event(order, ShipmentCreated(carrier_name=self.carrier_name))
        order.save()
        return order


# orders/apps.py — wire event policies at startup
import eventcore
from django.apps import AppConfig


class OrdersConfig(AppConfig):
    name = "apps.orders"

    def ready(self):
        from apps.notifications.tasks import SendShippingEmailTask
        from orders.events import ShipmentCreated

        eventcore.on(ShipmentCreated, lambda e: SendShippingEmailTask.delay(e.model.id))

Calling code — a view, a Celery task, the shell — only sees the command:

order = eventcore.invoke(ShipOrder(order_id=42, carrier_name="ups"))

The command validates the intent, invoke runs it in a transaction, the event mutates the model and announces the fact, and policies attach side effects without the command knowing about them. The rest of this README walks through each piece in detail.

Concepts

Concept	What it is
Command	A validated, named intention to change the system.
CommandError	An expected, domain-level failure raised from a command.
invoke(command)	Run a command in a transaction, then announce it to subscribers.
Event	A fact: something that happened (a pydantic payload).
ModelEvent[M]	An event that knows how to apply() itself to a model of type M.
apply_event(model, ev)	Apply an event to a model and announce it to subscribers.
on(Type, fn)	Subscribe fn to a command/event type (and its subclasses).

The flow

invoke(command)   ──applies──▶  apply_event(model, SomeEvent(...))
                                      │
                                      ├─▶ event.apply(model)   # mutate state in place
                                      └─▶ emit(event)          # notify subscribers
                                              │
                                              └─▶ on(SomeEvent) handlers run (after commit)

Defining events

from eventcore import ModelEvent
from orders.models import Order, Shipment

OrderEvent = ModelEvent[Order]
ShipmentEvent = ModelEvent[Shipment]


class ShipmentCreated(OrderEvent):
    carrier_name: str
    carrier_account_id: str

    def apply(self, model: Order) -> None:
        model.shipments.create(
            carrier_name=self.carrier_name,
            carrier_account_id=self.carrier_account_id,
        )


class ShipmentDispatched(ShipmentEvent):
    dispatcher: str

    def apply(self, model: Shipment) -> None:
        model.status = Shipment.Status.DISPATCHED
        model.dispatched_by = self.dispatcher
        # Occurrence metadata is available via self.meta (see below).
        model.dispatched_at = self.meta.occurred_at

Apply an event to a plain model with the apply_event free function — there is no base class or mixin to add:

from django.db import models
from eventcore import apply_event


class Order(models.Model):
    ...


apply_event(order, ShipmentCreated(carrier_name="ups", carrier_account_id="A1"))

apply_event mutates the instance and (by default) defers subscriber dispatch to transaction commit. It does not call save() — the calling command decides when to persist (see Convention, not enforcement). It must run inside a command invocation (see Occurrence metadata); calling it outside invoke() raises.

Defining commands

from eventcore import Command, CommandError, apply_event
from orders.models import Order
from orders.events import ShipmentCreated


class ShipOrder(Command):
    order_id: int
    carrier_name: str

    class OrderNotFound(CommandError): ...
    class CarrierNotConfigured(CommandError): ...

    def perform(self):
        try:
            order = Order.objects.select_related("store").get(
                pk=self.order_id
            )
        except Order.DoesNotExist:
            raise self.OrderNotFound(
                "Order %s not found", self.order_id
            )

        account_id = order.store.carrier_accounts.get(
            self.carrier_name
        )
        if account_id is None:
            raise self.CarrierNotConfigured(
                "Carrier %s not configured for store %s",
                self.carrier_name,
                order.store_id,
            )

        apply_event(
            order,
            ShipmentCreated(
                carrier_name=self.carrier_name,
                carrier_account_id=account_id,
            ),
        )
        order.save()
        return order

Run it:

order = eventcore.invoke(ShipOrder(order_id=1, carrier_name="ups"))

Subclasses implement perform(); you construct the command (pydantic validates the payload) and run it with the eventcore.invoke() free function. It runs perform() inside transaction.atomic() — always — so every event applied within the command commits atomically and their side effects (dispatched on_commit) fire only once the unit of work is durable. After the work commits, invoke() emits the command to on(Command, ...) subscribers (audit logs, metrics, …), so they react without the command knowing about them.

Occurrence metadata (meta)

invoke() establishes an EventMeta envelope for the run — occurred_at (now), actor (unset for now), and a fresh correlation_id — and makes it the active context (a contextvars.ContextVar) while perform() executes. Every apply_event call inside that command picks up the same envelope and binds it onto the event, so:

• an event's apply() can read self.meta.occurred_at (and actor / correlation_id), and so can any subscriber via event.meta;
• the command and all of its events share one correlation_id — the eventcore.app log records it on every row, so a command and its effects can be traced together.

Because the envelope comes from the command context, events can only be applied inside invoke() — apply_event raises otherwise. This is deliberate: every state change is attributable to a command. The envelope is captured onto the event at apply time, so on_commit subscribers still see event.meta even though the context has since unwound. (The context is per-thread and does not propagate into Celery tasks; pass any correlation_id you need explicitly.)

Reacting to events: policies

A policy is a rule that reacts to a domain event: when an event is applied, the policy runs to trigger follow-up actions (another command, a Celery task, a notification) or side effects, without the code that applied the event knowing about it. "Policy" is the name this codebase gives those event reactions; by convention an app defines them in a policies.py module:

# orders/policies.py
from orders.events import ShipmentCreated


def notify_customer(event: ShipmentCreated) -> None:
    from apps.notifications.tasks import SendShippingEmailTask

    SendShippingEmailTask.delay(event.model.id)

Wire them to their events in the app's apps.py ready(), with one eventcore.on(...) line per policy — so ready() is the single place that shows what reacts to what:

from django.apps import AppConfig


class OrdersConfig(AppConfig):
    name = "apps.orders"

    def ready(self):
        import eventcore

        from orders.events import ShipmentCreated
        from orders.policies import notify_customer

        eventcore.on(ShipmentCreated, notify_customer)

A policy registered for a base type also receives subclasses, so on(ModelEvent, audit) observes every model event.

Policies are also the only sanctioned way to chain commands. A command may not call invoke() from inside its own perform() — eventcore raises NestedCommandError if you try. To run command B in response to command A, react to one of A's events (or to A itself via on(Command, ...)) with a policy that invokes B. Because policies run after the first command commits, the second command starts as its own top-level transition with its own correlation context.

Commands flow through the same registry, so on(Command, ...) can observe every command invocation too (the optional logging app uses this). Such command listeners are plain handlers — the term policy is reserved for the event reactions above.

Dispatch timing

Subscribers run after the surrounding database transaction commits (transaction.on_commit), so side effects never fire for work that gets rolled back. In autocommit (no active transaction) they run immediately.

In tests, run the deferred handlers within a (rolled-back) test transaction using Django's on-commit capture — pytest-django exposes it as a fixture:

with django_capture_on_commit_callbacks(execute=True):
    invoke(PlaceOrder(customer_name="Ada", item="Keyboard"))
# subscribers have now run

The registry's dispatch/matching logic can also be tested directly and synchronously via Registry.run(message), without a transaction. eventcore.clear() removes all registered subscribers (useful for test isolation).

Convention, not enforcement

apply_event(model, event) mutates the model and emits the event — and nothing else. It does not override save(), intercept field assignment, or block queryset update(). Models are plain Django models with no base class or mixin, and the full ORM keeps working.

Routing state transitions through commands and events is a discipline the codebase adopts, not something the framework polices. That cuts both ways:

• Direct mutation bypasses subscribers. If code does order.status = "shipped"; order.save(), no event is emitted and nothing reacts — no notification task, no audit handler. That is fine for incidental bookkeeping (counters, denormalized caches, internal timestamps), and wrong for any transition other parts of the system care about.
• Persistence stays explicit. Because neither events nor apply_event ever call save(), a command can apply several events to one or more models and persist once, keeping the whole transition atomic alongside invoke()'s transaction.atomic() wrapper.

A useful rule of thumb: if a change is a domain fact — a status change, a lifecycle milestone, anything another team member would name in conversation — it deserves a command and an event. If it is plumbing no one would ever subscribe to, the plain ORM is the right tool.

Introspection

Every Command and Event subclass registers itself when its module is imported, so the running process can enumerate them:

from eventcore import registered_commands, registered_events

registered_commands()  # -> [CancelOrder, PlaceOrder, ShipOrder, ...]
registered_events()    # -> [OrderCancelled, OrderPlaced, OrderShipped, ...]

This makes a live catalog (names, pydantic fields, docstrings) available to, say, an API that documents the domain's commands and events.

Optional logging app

Both commands and events announce themselves through the registry after the work commits, so a subscriber can persist an audit trail. The bundled eventcore.app Django app does exactly that, subscribing with on(Command, …) and on(Event, …):

# settings.py
INSTALLED_APPS += ["eventcore.app"]

Installing it records a LogRecord row for every command invocation and every applied event. It is entirely opt-in: eventcore itself never imports the app, and the core command/event machinery works whether or not it is installed — with no app, the registry simply has no logging subscriber attached.

What this is not

• Not event sourcing — there is no event log and no rebuild-from-events.
• Not a message bus — dispatch is in-process; durable/async fan-out is the job of the subscribers (e.g. enqueue a Celery task).
• Not a persistence layer — events mutate models but never save().