Decorators

Protean provides decorators to help you construct elements of your domain model. Below is a sneak-preview into the various domain elements supported by Protean. Each element is explored in detail in its own section.

Domain.aggregate

from protean import Domain
from protean.fields import Integer, String

domain = Domain(__file__)


@domain.aggregate
class User:
    first_name = String(max_length=50)
    last_name = String(max_length=50)
    age = Integer()

Read more at Aggregates.

Domain.entity

from protean import Domain
from protean.fields import Integer, String

domain = Domain(__file__)


@domain.aggregate
class User:
    first_name = String(max_length=50)
    last_name = String(max_length=50)
    age = Integer()


@domain.entity(part_of=User)
class Credentials:
    email = String(max_length=254)
    password_hash = String(max_length=128)

Read more at Entities.

Domain.value_object

from protean import Domain
from protean.fields import Integer, String, ValueObject

domain = Domain(__file__)


@domain.value_object
class Address:
    address1 = String(max_length=255, required=True)
    address2 = String(max_length=255)
    address3 = String(max_length=255)
    city = String(max_length=25, required=True)
    state = String(max_length=25, required=True)
    country = String(max_length=2, required=True)
    zip = String(max_length=6, required=True)


@domain.aggregate
class User:
    first_name = String(max_length=50)
    last_name = String(max_length=50)
    age = Integer()
    address = ValueObject(Address)

Read more at Value Objects.

Domain.domain_service

from protean import Domain
from protean.fields import Identifier, Integer, String, ValueObject

domain = Domain(__file__)


@domain.aggregate
class User:
    first_name = String(max_length=50)
    last_name = String(max_length=50)
    age = Integer()


@domain.value_object(part_of="Subscription")
class Subscriber:
    id = Identifier()
    full_name = String(max_length=102)


@domain.aggregate
class Subscription:
    plan = String(max_length=50)
    user = ValueObject(Subscriber)
    status = String(max_length=50)


@domain.aggregate
class Plan:
    name = String(max_length=50)
    price = Integer()


@domain.domain_service
class SubscriptionManagement:
    def subscribe_user(self, user, plan):
        subscription = Subscription(user=user, plan=plan, status="ACTIVE")
        return subscription

Read more at Domain Services.

Domain.event_sourced_aggregate

from protean import Domain
from protean.fields import Integer, String

domain = Domain(__file__)


@domain.event_sourced_aggregate
class Person:
    name = String()
    age = Integer()

Read more at Event Sourced Aggregates.

Domain.command

from protean import Domain
from protean.fields import Identifier, String

domain = Domain(__file__)


@domain.aggregate
class User:
    name = String(max_length=50)


@domain.entity(part_of=User)
class Credentials:
    email = String(max_length=254)
    password_hash = String(max_length=128)


@domain.command(part_of=User)
class Register:
    id = Identifier()
    email = String()
    name = String()
    password_hash = String()

Read more at Commands.

Domain.command_handler

from protean import Domain, handle
from protean.fields import Identifier, String

domain = Domain(__file__)


@domain.event_sourced_aggregate
class User:
    id = Identifier()
    email = String()
    name = String()


@domain.command(part_of=User)
class Register:
    user_id = Identifier()
    email = String()


@domain.command(part_of=User)
class ChangePassword:
    old_password_hash = String()
    new_password_hash = String()


@domain.command_handler
class UserCommandHandlers:
    @handle(Register)
    def register(self, command: Register) -> None:
        pass

    @handle(ChangePassword)
    def change_password(self, command: ChangePassword) -> None:
        pass

Read more at Command Handlers.

Domain.event

from protean import Domain
from protean.fields import Identifier, String

domain = Domain(__file__)


@domain.aggregate
class User:
    name = String(max_length=50)


@domain.entity(part_of=User)
class Credentials:
    email = String(max_length=254)
    password_hash = String(max_length=128)


@domain.event(part_of=User)
class Registered:
    id = Identifier()
    email = String()
    name = String()
    password_hash = String()

Read more at Events.

Domain.event_handler

from protean import Domain, handle
from protean.fields import Identifier, String

domain = Domain(__file__)


@domain.aggregate
class User:
    id = Identifier()
    email = String()
    name = String()


@domain.command(part_of=User)
class Register:
    id = Identifier()
    email = String()


@domain.event(part_of=User)
class Registered:
    id = Identifier()
    email = String()
    name = String()
    password_hash = String()


@domain.event_handler
class UserEventHandlers:
    @handle(Registered)
    def send_email_notification(self, event: Registered) -> None:
        pass

Read more at Event Handlers.

Domain.model

import uuid

import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID

from protean import Domain
from protean.fields import Identifier, String

domain = Domain(__file__)


@domain.aggregate
class User:
    id = Identifier()
    email = String()
    name = String()


@domain.model(part_of=User)
class UserCustomModel:
    id = sa.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = sa.Column(sa.String(50))
    email = sa.Column(sa.String(254))

    class Meta:
        schema_name = "customers"

Read more at Models.

Domain.repository

from typing import List

from protean import Domain
from protean.fields import Integer, String
from protean.utils.globals import current_domain

domain = Domain(__file__)


@domain.aggregate
class Person:
    first_name = String(max_length=50, required=True)
    last_name = String(max_length=50, required=True)
    age = Integer(default=21)


@domain.repository(part_of=Person)
class PersonCustomRepository:
    def adults(self, minimum_age: int = 21) -> List[Person]:
        return current_domain.repository_for(Person)._dao.filter(age__gte=minimum_age)

Read more at Repositories.

Domain.view

from protean import Domain
from protean.fields import Identifier, Integer, String

domain = Domain(__file__)


@domain.aggregate
class User:
    first_name = String(max_length=50)
    last_name = String(max_length=50)
    age = Integer()


@domain.entity(part_of=User)
class Credentials:
    email = String(max_length=254)
    password_hash = String(max_length=128)


@domain.view(part_of=User)
class Token:
    key = Identifier(identifier=True)
    id = Identifier(required=True)
    email = String(required=True)

Read more at Views.