Skip to content

Quickstart

Build your first domain with Protean in 5 minutes. You will model a simple blog post system that creates posts via commands, publishes them, and reacts to events — all running in-memory with no infrastructure required.

Prerequisites

Create a Domain

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

Create a file called blog.py and add:

from protean import Domain

domain = Domain()

That's it. Protean provides in-memory adapters for databases, brokers, and event stores out of the box, so you can focus on your domain logic without setting up any infrastructure.

Define an Aggregate

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

Let's model a Post:

from protean import Domain, handle, invariant
from protean.exceptions import ValidationError
from protean.fields import Identifier, String, Text
from protean.utils.globals import current_domain

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))

    @invariant.post
    def body_must_be_substantial_when_published(self):
        if self.status == "PUBLISHED" and (not self.body or len(self.body) < 10):
            raise ValidationError(
                {"body": ["A post must have at least 10 characters to publish."]}
            )

A few things to note:

  • Fields like String and Text define the aggregate's data with built-in validation (max_length, required).
  • Methods like publish() encapsulate behavior and raise events when state changes.
  • Invariants enforce business rules that span multiple fields. Here, the @invariant.post decorator checks after every state change that a published post has a substantial body — something that field-level validation alone cannot express.

Define an Event

Events represent things that happened. They are named in past tense and are raised from within aggregates:

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

The part_of option connects the event to its aggregate. Events are immutable records — they capture facts about state changes.

Define a Command and Handler

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

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




@domain.command_handler(part_of=Post)
class PostCommandHandler:

The command handler follows a simple pattern: receive a command, create or load an aggregate, mutate state, and persist. Protean automatically wraps each handler method in a transaction.

React to Events

Event handlers process events after they occur — for side effects like sending notifications, syncing other aggregates, or logging:

        current_domain.repository_for(Post).add(post)
        return post.id

Event handlers are decoupled from the aggregate that raised the event. In production, they run asynchronously via the Protean server.

Put It All Together

Initialize the domain and run the full cycle — create a post via a command, then publish it:

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




if __name__ == "__main__":
    domain.init()

    domain.config["event_processing"] = "sync"
    domain.config["command_processing"] = "sync"

    with domain.domain_context():
        # Create a post via command
        post_id = domain.process(
            CreatePost(title="Hello, Protean!", body="My first domain.")
        )

        # Retrieve it from the repository
        post = domain.repository_for(Post).get(post_id)

Run it:

$ python blog.py
Post: Hello, Protean! (status: DRAFT)
Post published: Hello, Protean!
Updated: Hello, Protean! (status: PUBLISHED)

What Just Happened?

Here's the flow that Protean orchestrated for you:

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

    App->>Domain: Process CreatePost command
    Domain->>Handler: Dispatch command
    Handler->>Repo: Persist new Post
    Repo-->>Handler: OK
    Handler-->>App: Return post_id

    App->>Repo: Load Post and publish
    App->>Repo: Persist updated Post
    Repo->>EH: Deliver PostPublished event
    EH->>EH: Print notification
  1. domain.process() routes the CreatePost command to its handler.
  2. The handler creates a Post aggregate and persists it.
  3. You load the post, call publish(), which changes status and raises a PostPublished event.
  4. When you persist the updated post, the event is delivered to the PostEventHandler.

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

Full Source

Here is the complete example in a single file:

from protean import Domain, handle, invariant
from protean.exceptions import ValidationError
from protean.fields import Identifier, String, Text
from protean.utils.globals import current_domain

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))

    @invariant.post
    def body_must_be_substantial_when_published(self):
        if self.status == "PUBLISHED" and (not self.body or len(self.body) < 10):
            raise ValidationError(
                {"body": ["A post must have at least 10 characters to publish."]}
            )




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




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




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




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




if __name__ == "__main__":
    domain.init()

    domain.config["event_processing"] = "sync"
    domain.config["command_processing"] = "sync"

    with domain.domain_context():
        # Create a post via command
        post_id = domain.process(
            CreatePost(title="Hello, Protean!", body="My first domain.")
        )

        # Retrieve it from the repository
        post = domain.repository_for(Post).get(post_id)
        print(f"Post: {post.title} (status: {post.status})")

        # Publish the post — this raises a PostPublished event
        post.publish()
        domain.repository_for(Post).add(post)

        # Verify
        updated = domain.repository_for(Post).get(post_id)
        print(f"Updated: {updated.title} (status: {updated.status})")

Next Steps

  • Compose a Domain — learn about domain registration, initialization, and activation in depth.
  • Define Concepts — explore aggregates, entities, value objects, and fields.
  • Add Behavior — add validations, invariants, and domain services.
  • Configuration — connect real databases, brokers, and event stores.