Common Arguments

Applies to: DDD · CQRS · Event Sourcing

description

A long form description of the field. This value can be used by database adapters to provide additional context to a field.

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

domain = Domain()


@domain.aggregate
class Building:
    permit: List(
        content_type=String, description="Licences and Approvals", required=True
    )

required

Indicates if the field is required (must have a value). If True, the field is not allowed to be blank. Default is False.

from protean import Domain
from protean.fields import String

domain = Domain()


@domain.aggregate
class Person:
    name: String(required=True)

Leaving the field blank or not specifying a value will raise a ValidationError:

In [1]: p = Person()
...
ValidationError: {'name': ['is required']}

identifier

If True, the field is an identifier for the entity (a la primary key in RDBMS).

from protean import Domain
from protean.fields import String

domain = Domain()


@domain.aggregate
class Person:
    email: String(identifier=True)
    name: String(required=True)

The field is validated to be unique and non-blank:

In [1]: from protean.utils.reflection import declared_fields

In [2]: p = Person(email='john.doe@example.com', name='John Doe')

In [3]: declared_fields(p)["email"].identifier
Out[3]: True

In [4]: p = Person(name='John Doe')
...
ValidationError: {'email': ['is required']}

Aggregates and Entities need at least one field to be marked as an identifier. If you don’t specify one, Protean will automatically add a field called id to act as the primary identifier. This means that you don’t need to explicitly set identifier=True on any of your fields unless you want to override the default behavior or the name of the field.

Alternatively, you can use Identifier field type for primary identifier fields.

By default, Protean dynamically generates UUIDs as values of identifier fields unless explicitly provided. You can customize the type of value accepted with identity-strategy config parameter. More details are in Configuration section.

default

The default value for the field if no value is provided.

This can be a value or a callable object. If callable, the function will be called every time a new object is created.

from datetime import UTC, datetime

from protean.domain import Domain
from protean.fields import DateTime, String

publishing = Domain(__name__)


def utc_now():
    return datetime.now(UTC)


@publishing.aggregate
class Post:
    title: String(max_length=50)
    created_at: DateTime(default=utc_now)

In [1]: post = Post(title='Foo')

In [2]: post.to_dict()
Out[2]:
{'title': 'Foo',
 'created_at': '2024-05-09T00:58:10.781744+00:00',
 'id': '4f6b1fef-bc60-44c2-9ba6-6f844e0d31b0',
 '_version': -1}

Mutable object defaults

Protean copies a list or dict default for each instance, so default=[] and default={} are safe. The copy is shallow: objects nested inside the list or dict are still shared between instances. When the default holds mutable objects, wrap it in a callable that builds a new one each time.

For example, to give a List field a default list of topics, use a function:

from protean.domain import Domain
from protean.fields import List, String

domain = Domain(__name__)


def standard_topics():
    return ["Music", "Cinema", "Politics"]


@domain.aggregate
class Adult:
    name: String(max_length=255)
    topics: List(default=standard_topics)

Initializing an Adult aggregate will populate the defaults correctly:

In [1]: adult = Adult(name="John Doe")

In [2]: adult.to_dict()
Out[2]:
{'name': 'John Doe',
 'topics': ['Music', 'Cinema', 'Politics'],
 'id': '14381a6f-b62a-4135-a1d7-d50f68e2afba',
 '_version': -1}

Lambda expressions

You can use lambda expressions to specify an anonymous function:

import random

from protean.domain import Domain
from protean.fields import Integer

domain = Domain(__name__)

dice_sides = [4, 6, 8, 10, 12, 20]


@domain.aggregate
class Dice:
    sides: Integer(default=lambda: random.choice(dice_sides))

    def throw(self):
        return random.randint(1, self.sides)

In [1]: dice = Dice()

In [2]: dice.to_dict()
Out[2]: {'sides': 6, 'id': '0536ade5-f3a4-4e94-8139-8024756659a7', '_version': -1}

In [3]: dice.throw()
Out[3]: 3

This is a great option when you want to pass parameters to a function.

unique

Indicates if the field values must be unique within the repository. If True, this field's value is validated to be unique among all entities.

from protean import Domain
from protean.fields import String

domain = Domain()


@domain.aggregate
class Person:
    name: String(required=True)
    email: String(unique=True)

Obviously, this field's integrity is enforced at the database layer when an entity is persisted. If an entity instance specifies a duplicate value in a field marked unique, a ValidationError will be raised:

In [1]: p1 = Person(name='John Doe', email='john.doe@example.com')

In [2]: domain.repository_for(Person).add(p1)
Out[2]: <Person: Person object (id: b2c592d5-bd78-4e1e-a9d1-eea20ab5374a)>

In [3]: p2 = Person(name= 'Jane Doe', email='john.doe@example.com')

In [4]: domain.repository_for(Person).add(p2)
ERROR: Failed saving entity because {'email': ["Person with email 'john.doe@example.com' is already present."]}
...
ValidationError: {'email': ["Person with email 'john.doe@example.com' is already present."]}

We will explore more about persistence in Application Layer guide.

choices

A set of allowed choices for the field value. When supplied as an Enum, the value of the field is validated to be one among the specified options.

from enum import Enum

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

domain = Domain()


class BuildingStatus(Enum):
    WIP = "WIP"
    DONE = "DONE"


@domain.aggregate
class Building:
    name: String(max_length=50)
    floors: Integer()
    status: String(choices=BuildingStatus)

The choices are enforced when the field is initialized or updated:

In [1]: building = Building(name="Atlantis", floors=3, status="WIP")

In [2]: building.to_dict()
Out[2]:
{'name': 'Atlantis',
 'floors': 3,
 'status': 'WIP',
 'id': 'c803c763-32d7-403f-b432-8835a258430e',
 '_version': -1}

In [3]: building.status = "COMPLETED"
...
ValidationError: {'status': ["Input should be 'WIP' or 'DONE'"]}

referenced_as

The name of the field as referenced in the database or external systems. Defaults to the field's name.

from protean import Domain
from protean.fields import String

domain = Domain()


@domain.aggregate
class Person:
    email: String(unique=True)
    name: String(referenced_as="fullname", required=True)

Protean will now persist the value under fullname instead of name.

In [1]: from protean.utils.reflection import declared_fields, attributes

In [2]: list(declared_fields(Person))
Out[2]: ['email', 'name', 'id']

In [3]: list(attributes(Person))
Out[3]: ['email', 'fullname', 'id', '_version']

validators

Additional validators to apply to the field value.

Validators are callable Class instances that are invoked whenever a field's value is changed. Protean's String field, for example, has two default validators: MinLengthValidator and MaxLengthValidator classes associated with min_length and max_length attributes.

from typing import Any

from protean import Domain
from protean.exceptions import ValidationError
from protean.fields import String

domain = Domain()


class EmailDomainValidator:
    def __init__(self, domain="example.com"):
        self.domain = domain
        self.message = f"Email does not belong to {self.domain}"

    def __call__(self, value: str) -> Any:
        if not value.endswith(self.domain):
            raise ValidationError(self.message)


@domain.aggregate
class Employee:
    email: String(identifier=True, validators=[EmailDomainValidator("mydomain.com")])

If the value fails to satisfy the validation, a ValidationError will be thrown with the custom error message.

In [1]: e = Employee(email="john@mydomain.com")

In [2]: e.to_dict()
Out[2]: {'email': 'john@mydomain.com', '_version': -1}

In [3]: e2 = Employee(email="john@otherdomain.com")
...
ValidationError: {'email': ['Email does not belong to mydomain.com']}

error_messages

A dictionary of custom error messages, keyed by error kind such as required, invalid, unique, or invalid_choice. The field accepts the argument and records it, but the simple fields in protean.fields do not use it yet: a validation error still reports the default message.

from protean import Domain
from protean.fields import Integer

domain = Domain()


@domain.aggregate
class Building:
    doors: Integer(
        required=True, error_messages={"required": "Every building needs some!"}
    )

Leaving doors out still raises the default message, not the custom one:

In [1]: Building()
...
ValidationError: {'doors': ['is required']}

renamed_from

The old name (or a list of old names) this field was renamed from. Declaring a rename lets a stored payload written under the old key deserialize into the renamed field without an upcaster.

@domain.aggregate
class Order:
    order_id = Identifier(identifier=True)


@domain.event(part_of=Order)
class OrderPlaced:
    order_id = Identifier(identifier=True)
    customer_name = String(renamed_from="name")  # single old name
    total = Float(renamed_from=["amount", "sum"])  # or a list of aliases

A stored OrderPlaced payload carrying the old name key loads into customer_name; the current name always wins if both are present, and any stale alias keys are dropped. The rename is also emitted into the IR, so protean ir diff reports a single safe field_renamed change instead of a breaking remove + add, and the generated Avro schema carries an aliases entry so an external Avro consumer resolves the old name on the wire. See Evolving Events Over Time for the end-to-end workflow and Event Versioning and Evolution for the strategies behind it.