Skip to content

Quickstart

Build your first Protean domain in a few minutes. You will model a blog Post: a command publishes a post, an event handler reacts to the publish, a projector keeps a read-optimized feed, and you query that feed to read the post back. Everything runs in-memory, with no infrastructure to set up.

Prerequisites

Create a domain

Every Protean application starts with a Domain, the container for your business logic.

Create a file called blog.py and add:

from protean import Domain, current_domain, handle
from protean.core.projector import on
from protean.fields import Identifier, String, Text

domain = Domain()

Protean ships in-memory adapters for databases, brokers, and event stores, so you can write your domain logic before you pick any infrastructure.

Define an aggregate

Aggregates are the core building blocks. They hold state and enforce business rules.

Here is the Post:

@domain.aggregate
class Post:
    title: String(max_length=100, required=True)
    body: Text(required=True)
    status: String(max_length=20, default="DRAFT")

    def publish(self):
        self.status = "PUBLISHED"
        self.raise_(PostPublished(post_id=self.id, title=self.title))

String and Text are fields. They declare the aggregate's data and validate it with options like max_length and required. The publish() method changes the post's status and raises an event to record what happened.

Define an event

Events record things that happened. They are named in the past tense and are raised from inside aggregates:

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

The part_of option connects the event to its aggregate. PostPublished carries the post's id and title, the facts a reader of the event needs.

Define a command and handler

Commands carry the intent to change state. They are named as imperative verbs. A command handler receives the command and makes the change:

@domain.command(part_of=Post)
class PublishPost:
    title: String(max_length=100, required=True)
    body: Text(required=True)


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

PostCommandHandler creates a Post from the command, calls publish(), and saves it through the repository. Protean wraps each handler method in a transaction. domain.process() routes a PublishPost command to this handler.

React to events

An event handler runs after an event, for side effects like sending a notification or updating another part of the system:

@domain.event_handler(part_of=Post)
class PostEventHandler:
    @handle(PostPublished)
    def announce(self, event: PostPublished):
        print(f"Event handled: post published ({event.title})")

PostEventHandler prints a line when a post is published. It is decoupled from the aggregate that raised the event. In production it runs asynchronously through the Protean server.

Build a read model

A projection is a read-optimized view, kept current by a projector that reacts to events:

@domain.projection
class PublishedPostsFeed:
    """A read-optimized feed of published posts."""

    post_id: Identifier(identifier=True, required=True)
    title: String(max_length=100, required=True)


@domain.projector(projector_for=PublishedPostsFeed, aggregates=[Post])
class PublishedPostsFeedProjector:
    """Maintains the PublishedPostsFeed projection from Post events."""

    @on(PostPublished)
    def on_post_published(self, event: PostPublished):
        feed_entry = PublishedPostsFeed(post_id=event.post_id, title=event.title)
        current_domain.repository_for(PublishedPostsFeed).add(feed_entry)

PublishedPostsFeed holds one row per published post. The projector listens for PostPublished and adds a row, so a query against the feed returns published posts without loading the Post aggregate.

Put it all together

Initialize the domain and run the full arc. The command publishes a post, and the last lines query the feed and print what the projector recorded:

if __name__ == "__main__":
    domain.config["command_processing"] = "sync"
    domain.config["event_processing"] = "sync"

    domain.init(traverse=False)

    with domain.domain_context():
        # Write: publish a post through the command.
        post_id = domain.process(
            PublishPost(title="Hello, Protean!", body="My first published post.")
        )
        post = domain.repository_for(Post).get(post_id)
        print(f"Post created: {post.title} (status: {post.status})")

        # Read: the projector has already filled the feed inline.
        feed = domain.view_for(PublishedPostsFeed).query.all()
        print(f"Published posts feed: {feed.total} row(s)")
        for entry in feed.items:
            print(f"  - {entry.title}")

Run it:

$ python blog.py
Event handled: post published (Hello, Protean!)
Post created: Hello, Protean! (status: PUBLISHED)
Published posts feed: 1 row(s)
  - Hello, Protean!

What just happened?

Here is the flow that Protean ran for you:

sequenceDiagram
    autonumber
    participant App
    participant Domain
    participant Handler as Command Handler
    participant Repo as Repository
    participant EH as Event Handler
    participant Proj as Projector

    App->>Domain: Process PublishPost command
    Domain->>Handler: Dispatch command
    Handler->>Repo: Create and publish Post, then persist
    Repo->>EH: Deliver PostPublished event
    Repo->>Proj: Deliver PostPublished event
    Proj->>Repo: Add a row to PublishedPostsFeed
    Handler-->>App: Return post_id
    App->>Repo: Query PublishedPostsFeed
    Repo-->>App: The published post
  1. domain.process() routes the PublishPost command to PostCommandHandler.
  2. The handler creates a Post, calls publish(), which changes the status and raises a PostPublished event, and persists the post.
  3. On commit, PostPublished reaches PostEventHandler, which prints the announcement, and PublishedPostsFeedProjector, which adds a row to PublishedPostsFeed.
  4. You query PublishedPostsFeed and get the published post back.

All of this runs in-memory, with no database, message broker, or event store. When you are ready for production, swap in real adapters with configuration.

Full source

Here is the complete example in a single file:

from protean import Domain, current_domain, handle
from protean.core.projector import on
from protean.fields import Identifier, String, Text

domain = Domain()


@domain.aggregate
class Post:
    title: String(max_length=100, required=True)
    body: Text(required=True)
    status: String(max_length=20, default="DRAFT")

    def publish(self):
        self.status = "PUBLISHED"
        self.raise_(PostPublished(post_id=self.id, title=self.title))


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


@domain.command(part_of=Post)
class PublishPost:
    title: String(max_length=100, required=True)
    body: Text(required=True)


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


@domain.event_handler(part_of=Post)
class PostEventHandler:
    @handle(PostPublished)
    def announce(self, event: PostPublished):
        print(f"Event handled: post published ({event.title})")


@domain.projection
class PublishedPostsFeed:
    """A read-optimized feed of published posts."""

    post_id: Identifier(identifier=True, required=True)
    title: String(max_length=100, required=True)


@domain.projector(projector_for=PublishedPostsFeed, aggregates=[Post])
class PublishedPostsFeedProjector:
    """Maintains the PublishedPostsFeed projection from Post events."""

    @on(PostPublished)
    def on_post_published(self, event: PostPublished):
        feed_entry = PublishedPostsFeed(post_id=event.post_id, title=event.title)
        current_domain.repository_for(PublishedPostsFeed).add(feed_entry)


if __name__ == "__main__":
    domain.config["command_processing"] = "sync"
    domain.config["event_processing"] = "sync"

    domain.init(traverse=False)

    with domain.domain_context():
        # Write: publish a post through the command.
        post_id = domain.process(
            PublishPost(title="Hello, Protean!", body="My first published post.")
        )
        post = domain.repository_for(Post).get(post_id)
        print(f"Post created: {post.title} (status: {post.status})")

        # Read: the projector has already filled the feed inline.
        feed = domain.view_for(PublishedPostsFeed).query.all()
        print(f"Published posts feed: {feed.total} row(s)")
        for entry in feed.items:
            print(f"  - {entry.title}")

Where to go next