Protean

Protean is an opinionated Python framework for building event-driven applications with Domain-Driven Design — aggregates, CQRS, and event sourcing are first-class, and your domain logic stays independent of the database, broker, and API you run it on.

Installation

Protean is available on PyPI:

$ pip install protean

Protean officially supports Python 3.11+.

Quick Start

A command flows to its handler, the aggregate raises an event, and an event handler reacts — all wired by the domain, independent of infrastructure:

from protean import Domain
from protean.fields import Boolean, Identifier, String
from protean.utils.mixins import handle

domain = Domain(name="Publishing")
domain.config["command_processing"] = "sync"
domain.config["event_processing"] = "sync"


@domain.event(part_of="Post")
class PostPublished:
    post_id = Identifier()
    title = String()


@domain.aggregate
class Post:
    title = String(required=True, max_length=200)
    is_published = Boolean(default=False)

    def publish(self):
        self.is_published = True
        self.raise_(PostPublished(post_id=self.id, title=self.title))


@domain.command(part_of="Post")
class PublishPost:
    post_id = Identifier(identifier=True)
    title = String()


@domain.command_handler(part_of="Post")
class PostCommandHandler:
    @handle(PublishPost)
    def publish(self, command):
        post = Post(id=command.post_id, title=command.title)
        post.publish()
        domain.repository_for(Post).add(post)


@domain.event_handler(part_of="Post")
class Notifications:
    @handle(PostPublished)
    def announce(self, event):
        print(f"Published: {event.title}")


domain.init(traverse=False)
with domain.domain_context():
    domain.process(PublishPost(post_id="1", title="Hello, Protean"))