flask_boilerplate.domain.primitives package

Submodules

flask_boilerplate.domain.primitives.aggregate_root module

Module defining the base class for aggregate roots in the domain layer.

An aggregate root is a cluster of domain objects that can be treated as a single unit. It is responsible for maintaining consistency and enforcing invariants within the aggregate. The aggregate root is the only entry point for modifying the aggregate.

This module provides the base class AggregateRoot that can be inherited to create custom aggregate roots. Aggregate roots typically contain entities, value objects, and domain events.

class flask_boilerplate.domain.primitives.aggregate_root.AggregateRoot

Bases: object

Base class for aggregate roots in the domain layer.

An aggregate root is a cluster of domain objects that can be treated as a single unit. It is responsible for maintaining consistency and enforcing invariants within the aggregate. The aggregate root is the only entry point for modifying the aggregate.

domain_events

A list of domain events raised by the aggregate root.

Type:

list[DomainEvent]

__eq__(other: Any) → bool

Compare two aggregate roots for equality.

Two aggregate roots are considered equal if they are of the same type and have the same unique identifier.

Parameters:

other (Any) – The object to compare with.

Returns:

True if the objects are equal, False otherwise.

Return type:

bool

__hash__() → int

Compute the hash value of the aggregate root.

The hash value is computed based on the unique identifier of the aggregate root.

Returns:

The hash value of the aggregate root.

Return type:

int

add_domain_event(event: DomainEvent) → None

Add a domain event to the aggregate root.

Domain events represent significant changes or actions that occur within the aggregate. They are typically used to decouple domain logic and enable event-driven architectures.

Parameters:

event (DomainEvent) – The domain event to add.

clear_domain_events() → None

Clear all domain events from the aggregate root.

This method is typically called after the domain events have been processed.

domain_events: list[DomainEvent]

flask_boilerplate.domain.primitives.domain_service module

Module defining the base class for domain services in the domain layer.

Domain services encapsulate business logic that does not naturally fit within an entity or value object. They are stateless and operate on domain objects to perform complex operations or enforce business rules.

This module provides the base class DomainService that can be inherited to create custom domain services. Domain services are typically used to orchestrate interactions between multiple domain objects or external systems.

class flask_boilerplate.domain.primitives.domain_service.DomainService

Bases: ABC, Generic[T]

Base class for domain services in the domain layer.

A domain service encapsulates business logic that operates on domain objects. It is stateless and provides methods to perform complex operations or enforce business rules.

Subclasses should implement the methods to provide specific business logic.

Example

>>> class PaymentService(DomainService[Order]):
...     def process_payment(self, order: Order) -> bool:
...         # Business logic for processing payment
...         pass

abstractmethod execute(entity: T) → Any

Execute the domain service logic on the given entity.

This method encapsulates the core business logic of the domain service. It operates on the provided entity and returns the result of the operation.

Parameters:

entity (T) – The domain object to operate on.

Returns:

The result of the domain service operation.

Return type:

Any

flask_boilerplate.domain.primitives.entity module

class flask_boilerplate.domain.primitives.entity.Entity(id: UUID | None = None)

Bases: ABC

Base class for all entities in the domain layer.

An entity is an object that is defined by its identity rather than its attributes. Entities have a unique identifier and can be compared based on this identifier.

id

The unique identifier of the entity.

Type:

uuid.UUID

abstractmethod __eq__(other: Any) → bool

Compare two entities based on their unique identifier.

Parameters:

other (Entity) – The other entity to compare with.

Returns:

True if the entities have the same identifier, False otherwise.

Return type:

bool

abstractmethod __hash__() → int

Generate a hash value for the entity based on its unique identifier.

Returns:

The hash value of the entity.

Return type:

int

__init__(id: UUID | None = None) → None

Initialize a new entity with a unique identifier.

Parameters:

id (Optional[uuid.UUID]) – The unique identifier of the entity. If not provided, a new UUID is generated.

id: UUID

flask_boilerplate.domain.primitives.interface_domain_event module

Module defining the base interface for domain events in the domain layer.

Domain events are used to represent significant changes or actions that occur within the domain. They are typically used to decouple domain logic and enable event-driven architectures.

This module provides the base interface DomainEvent that can be implemented to create custom domain events. Domain events are immutable and should contain all the information necessary to describe the event.

class flask_boilerplate.domain.primitives.interface_domain_event.DomainEvent

Bases: ABC

Base interface for domain events in the domain layer.

A domain event represents a significant change or action that occurs within the domain. It is immutable and contains all the information necessary to describe the event.

Subclasses should implement the event_name property and any additional attributes required to describe the event.

Example

>>> @dataclass(frozen=True)
>>> class UserRegistered(DomainEvent):
...     user_id: str
...     email: str
...     timestamp: datetime = datetime.now()
...
...     @property
...     def event_name(self) -> str:
...         return "user_registered"

abstract property event_name: str

Get the name of the domain event.

The event name is used to identify the type of event and is typically used for routing or logging purposes.

Returns:

The name of the domain event.

Return type:

str

abstract property timestamp: datetime

Get the timestamp of when the domain event occurred.

Returns:

The timestamp of the domain event.

Return type:

datetime

class flask_boilerplate.domain.primitives.interface_domain_event.DomainEventBase(timestamp: datetime = <property object>)

Bases: DomainEvent

Base implementation of the DomainEvent interface.

This class provides a default implementation for the timestamp property and can be used as a base class for custom domain events.

Example

>>> @dataclass(frozen=True)
>>> class UserRegistered(DomainEventBase):
...     user_id: str
...     email: str
...
...     @property
...     def event_name(self) -> str:
...         return "user_registered"

property event_name: str

Get the name of the domain event.

This method must be implemented by subclasses to provide the event name.

Returns:

The name of the domain event.

Return type:

str

flask_boilerplate.domain.primitives.repository module

Module defining the base interfaces for repositories in the domain layer.

Repositories are used to abstract the persistence and retrieval of domain objects. They provide a collection-like interface for accessing domain objects while hiding the details of the underlying data storage.

This module provides the base interfaces Repository and UnitOfWork that can be implemented to create custom repositories and transaction management mechanisms.

class flask_boilerplate.domain.primitives.repository.Repository

Bases: Generic[T, ID], ABC

Base interface for repositories in the domain layer.

A repository provides a collection-like interface for accessing domain objects. It abstracts the details of data storage and retrieval, allowing the domain layer to remain agnostic of the underlying persistence mechanism.

Subclasses should implement the methods to provide specific persistence logic.

Example

>>> class UserRepository(Repository[User, UUID]):
...     def add(self, user: User) -> None:
...         pass
...
...     def get(self, user_id: UUID) -> Optional[User]:
...         pass
...
...     def list(self) -> Iterable[User]:
...         pass
...
...     def remove(self, user: User) -> None:
...         pass

abstractmethod add(entity: T) → None

Add a new entity to the repository.

Parameters:

entity (T) – The entity to add.

abstractmethod get(id: ID) → T | None

Retrieve an entity by its unique identifier.

Parameters:

id (ID) – The unique identifier of the entity.

Returns:

The entity if found, otherwise None.

Return type:

Optional[T]

abstractmethod list() → Iterable[T]

Retrieve all entities from the repository.

Returns:

A collection of all entities.

Return type:

Iterable[T]

abstractmethod remove(entity: T) → None

Remove an entity from the repository.

Parameters:

entity (T) – The entity to remove.

class flask_boilerplate.domain.primitives.repository.UnitOfWork

Bases: Generic[T, ID], ABC

Base interface for the Unit of Work pattern in the domain layer.

The Unit of Work pattern is used to manage transactions and track changes to domain objects. It ensures that all changes are committed or rolled back as a single unit.

Subclasses should implement the methods to provide specific transaction management.

Example

>>> class UserUnitOfWork(UnitOfWork[User, UUID]):
...     def commit(self) -> None:
...         pass
...
...     def rollback(self) -> None:
...         pass
...
...     def repositories(self) -> dict[str, Repository[User, UUID]]:
...         pass

abstractmethod commit() → None

Commit all changes made within the unit of work.

abstractmethod repositories() → dict[str, Repository[T, ID]]

Get the repositories managed by this unit of work.

Returns:

A dictionary of repositories, keyed by name.

Return type:

dict[str, Repository[T, ID]]

abstractmethod rollback() → None

Roll back all changes made within the unit of work.

flask_boilerplate.domain.primitives.specification module

Module defining the base class for specifications in the domain layer.

Specifications are used to encapsulate business rules or conditions that can be used to filter, query, or validate domain objects. They follow the Specification pattern, which allows for reusable and composable business logic.

This module provides a base class Specification that can be inherited to create custom specifications. Specifications can be combined using logical operators (&, |, ~) to create complex conditions.

class flask_boilerplate.domain.primitives.specification.AndSpecification(first: Specification[T], second: Specification[T])

Bases: Specification[T]

Specification representing the logical AND of two specifications.

__init__(first: Specification[T], second: Specification[T]) → None

Initialize the AndSpecification.

Parameters:

• first (Specification[T]) – The first specification.

• second (Specification[T]) – The second specification.

is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies both specifications.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies both specifications, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.specification.NotSpecification(specification: Specification[T])

Bases: Specification[T]

Specification representing the logical NOT of a specification.

__init__(specification: Specification[T]) → None

Initialize the NotSpecification.

Parameters:

specification (Specification[T]) – The specification to negate.

is_satisfied_by(candidate: T) → bool

Check if the candidate does not satisfy the specification.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate does not satisfy the specification, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.specification.OrSpecification(first: Specification[T], second: Specification[T])

Bases: Specification[T]

Specification representing the logical OR of two specifications.

__init__(first: Specification[T], second: Specification[T]) → None

Initialize the OrSpecification.

Parameters:

• first (Specification[T]) – The first specification.

• second (Specification[T]) – The second specification.

is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies at least one of the specifications.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies at least one specification, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.specification.Specification

Bases: ABC, Generic[T]

Base class for specifications in the domain layer.

A specification encapsulates a business rule or condition that can be evaluated against a domain object. It follows the Specification pattern, allowing for reusable and composable business logic.

Subclasses should implement the is_satisfied_by method to define the specific condition or rule.

Example

>>> class IsActiveSpecification(Specification[User]):
...     def is_satisfied_by(self, user: User) -> bool:
...         return user.is_active
...
>>> active_spec = IsActiveSpecification()
>>> user = User(is_active=True)
>>> active_spec.is_satisfied_by(user)
True

__and__(other: Specification[T]) → AndSpecification[T]

Combine two specifications using the logical AND operator.

Parameters:

other (Specification[T]) – The other specification to combine with.

Returns:

A new specification representing the logical AND.

Return type:

AndSpecification[T]

__invert__() → NotSpecification[T]

Negate the specification using the logical NOT operator.

Returns:

A new specification representing the logical NOT.

Return type:

NotSpecification[T]

__or__(other: Specification[T]) → OrSpecification[T]

Combine two specifications using the logical OR operator.

Parameters:

other (Specification[T]) – The other specification to combine with.

Returns:

A new specification representing the logical OR.

Return type:

OrSpecification[T]

abstractmethod is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies the specification.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies the specification, False otherwise.

Return type:

bool

flask_boilerplate.domain.primitives.value_object module

Module defining the base class for value objects in the domain layer.

Value objects are immutable objects that are defined by their attributes rather than a unique identity. They encapsulate domain-specific logic and ensure consistency and validity of data within the domain.

This module provides a base class ValueObject that can be inherited to create custom value objects. Value objects are compared based on their attributes rather than their identity.

class flask_boilerplate.domain.primitives.value_object.ValueObject

Bases: ABC

Abstract base class for all value objects in the domain layer.

A value object is an immutable object that is defined by its attributes. It does not have a unique identity and is compared based on its attribute values.

Implements core value object characteristics: - Immutability - Structural equality - Validation - Serialization/deserialization

Subclasses should define their attributes using dataclass fields. The frozen=True parameter ensures immutability.

Inheriting classes must implement: - _validate() method - to_primitives() method

Example

>>> @dataclass(frozen=True)
>>> class Money(ValueObject):
...     amount: float
...     currency: str
...
>>> money1 = Money(amount=100.0, currency="USD")
>>> money2 = Money(amount=100.0, currency="USD")
>>> money1 == money2
True

abstractmethod __eq__(other: Any) → bool

Compare two value objects based on their attributes. :param other: The other value object to compare with. :type other: Any

Returns:

True if the value objects have the same attributes, False otherwise.

Return type:

bool

abstractmethod __hash__() → int

Generate a hash value for the value object based on its attributes. :returns: The hash value of the value object. :rtype: int

__post_init__() → None

Post-initialization hook to ensure the value object is properly validated.

validate() → None

Validate the value object’s data. :raises ValueObjectsError: If the value object’s data is invalid.

Module contents

Module for exporting primitive domain constructs.

This module serves as the entry point for all primitive domain constructs, such as value objects, specifications, repositories, domain events, and aggregate roots. It re-exports these constructs to make them easily accessible from a single location.

Example

>>> from flask_boilerplate.domain.primitives import ValueObject, Specification, Repository
>>> class MyValueObject(ValueObject):
...     pass

class flask_boilerplate.domain.primitives.AggregateRoot

Bases: object

Base class for aggregate roots in the domain layer.

An aggregate root is a cluster of domain objects that can be treated as a single unit. It is responsible for maintaining consistency and enforcing invariants within the aggregate. The aggregate root is the only entry point for modifying the aggregate.

domain_events

A list of domain events raised by the aggregate root.

Type:

list[DomainEvent]

__eq__(other: Any) → bool

Compare two aggregate roots for equality.

Two aggregate roots are considered equal if they are of the same type and have the same unique identifier.

Parameters:

other (Any) – The object to compare with.

Returns:

True if the objects are equal, False otherwise.

Return type:

bool

__hash__() → int

Compute the hash value of the aggregate root.

The hash value is computed based on the unique identifier of the aggregate root.

Returns:

The hash value of the aggregate root.

Return type:

int

add_domain_event(event: DomainEvent) → None

Add a domain event to the aggregate root.

Domain events represent significant changes or actions that occur within the aggregate. They are typically used to decouple domain logic and enable event-driven architectures.

Parameters:

event (DomainEvent) – The domain event to add.

clear_domain_events() → None

Clear all domain events from the aggregate root.

This method is typically called after the domain events have been processed.

domain_events: list[DomainEvent]

class flask_boilerplate.domain.primitives.AndSpecification(first: Specification[T], second: Specification[T])

Bases: Specification[T]

Specification representing the logical AND of two specifications.

__init__(first: Specification[T], second: Specification[T]) → None

Initialize the AndSpecification.

Parameters:

• first (Specification[T]) – The first specification.

• second (Specification[T]) – The second specification.

is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies both specifications.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies both specifications, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.DomainEvent

Bases: ABC

Base interface for domain events in the domain layer.

A domain event represents a significant change or action that occurs within the domain. It is immutable and contains all the information necessary to describe the event.

Subclasses should implement the event_name property and any additional attributes required to describe the event.

Example

>>> @dataclass(frozen=True)
>>> class UserRegistered(DomainEvent):
...     user_id: str
...     email: str
...     timestamp: datetime = datetime.now()
...
...     @property
...     def event_name(self) -> str:
...         return "user_registered"

abstract property event_name: str

Get the name of the domain event.

The event name is used to identify the type of event and is typically used for routing or logging purposes.

Returns:

The name of the domain event.

Return type:

str

abstract property timestamp: datetime

Get the timestamp of when the domain event occurred.

Returns:

The timestamp of the domain event.

Return type:

datetime

class flask_boilerplate.domain.primitives.DomainEventBase(timestamp: datetime = <property object>)

Bases: DomainEvent

Base implementation of the DomainEvent interface.

This class provides a default implementation for the timestamp property and can be used as a base class for custom domain events.

Example

>>> @dataclass(frozen=True)
>>> class UserRegistered(DomainEventBase):
...     user_id: str
...     email: str
...
...     @property
...     def event_name(self) -> str:
...         return "user_registered"

property event_name: str

Get the name of the domain event.

This method must be implemented by subclasses to provide the event name.

Returns:

The name of the domain event.

Return type:

str

class flask_boilerplate.domain.primitives.DomainService

Bases: ABC, Generic[T]

Base class for domain services in the domain layer.

A domain service encapsulates business logic that operates on domain objects. It is stateless and provides methods to perform complex operations or enforce business rules.

Subclasses should implement the methods to provide specific business logic.

Example

>>> class PaymentService(DomainService[Order]):
...     def process_payment(self, order: Order) -> bool:
...         # Business logic for processing payment
...         pass

abstractmethod execute(entity: T) → Any

Execute the domain service logic on the given entity.

This method encapsulates the core business logic of the domain service. It operates on the provided entity and returns the result of the operation.

Parameters:

entity (T) – The domain object to operate on.

Returns:

The result of the domain service operation.

Return type:

Any

class flask_boilerplate.domain.primitives.NotSpecification(specification: Specification[T])

Bases: Specification[T]

Specification representing the logical NOT of a specification.

__init__(specification: Specification[T]) → None

Initialize the NotSpecification.

Parameters:

specification (Specification[T]) – The specification to negate.

is_satisfied_by(candidate: T) → bool

Check if the candidate does not satisfy the specification.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate does not satisfy the specification, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.OrSpecification(first: Specification[T], second: Specification[T])

Bases: Specification[T]

Specification representing the logical OR of two specifications.

__init__(first: Specification[T], second: Specification[T]) → None

Initialize the OrSpecification.

Parameters:

• first (Specification[T]) – The first specification.

• second (Specification[T]) – The second specification.

is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies at least one of the specifications.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies at least one specification, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.Repository

Bases: Generic[T, ID], ABC

Base interface for repositories in the domain layer.

A repository provides a collection-like interface for accessing domain objects. It abstracts the details of data storage and retrieval, allowing the domain layer to remain agnostic of the underlying persistence mechanism.

Subclasses should implement the methods to provide specific persistence logic.

Example

>>> class UserRepository(Repository[User, UUID]):
...     def add(self, user: User) -> None:
...         pass
...
...     def get(self, user_id: UUID) -> Optional[User]:
...         pass
...
...     def list(self) -> Iterable[User]:
...         pass
...
...     def remove(self, user: User) -> None:
...         pass

abstractmethod add(entity: T) → None

Add a new entity to the repository.

Parameters:

entity (T) – The entity to add.

abstractmethod get(id: ID) → T | None

Retrieve an entity by its unique identifier.

Parameters:

id (ID) – The unique identifier of the entity.

Returns:

The entity if found, otherwise None.

Return type:

Optional[T]

abstractmethod list() → Iterable[T]

Retrieve all entities from the repository.

Returns:

A collection of all entities.

Return type:

Iterable[T]

abstractmethod remove(entity: T) → None

Remove an entity from the repository.

Parameters:

entity (T) – The entity to remove.

class flask_boilerplate.domain.primitives.Specification

Bases: ABC, Generic[T]

Base class for specifications in the domain layer.

A specification encapsulates a business rule or condition that can be evaluated against a domain object. It follows the Specification pattern, allowing for reusable and composable business logic.

Subclasses should implement the is_satisfied_by method to define the specific condition or rule.

Example

>>> class IsActiveSpecification(Specification[User]):
...     def is_satisfied_by(self, user: User) -> bool:
...         return user.is_active
...
>>> active_spec = IsActiveSpecification()
>>> user = User(is_active=True)
>>> active_spec.is_satisfied_by(user)
True

__and__(other: Specification[T]) → AndSpecification[T]

Combine two specifications using the logical AND operator.

Parameters:

other (Specification[T]) – The other specification to combine with.

Returns:

A new specification representing the logical AND.

Return type:

AndSpecification[T]

__invert__() → NotSpecification[T]

Negate the specification using the logical NOT operator.

Returns:

A new specification representing the logical NOT.

Return type:

NotSpecification[T]

__or__(other: Specification[T]) → OrSpecification[T]

Combine two specifications using the logical OR operator.

Parameters:

other (Specification[T]) – The other specification to combine with.

Returns:

A new specification representing the logical OR.

Return type:

OrSpecification[T]

abstractmethod is_satisfied_by(candidate: T) → bool

Check if the candidate satisfies the specification.

Parameters:

candidate (T) – The domain object to evaluate.

Returns:

True if the candidate satisfies the specification, False otherwise.

Return type:

bool

class flask_boilerplate.domain.primitives.UnitOfWork

Bases: Generic[T, ID], ABC

Base interface for the Unit of Work pattern in the domain layer.

The Unit of Work pattern is used to manage transactions and track changes to domain objects. It ensures that all changes are committed or rolled back as a single unit.

Subclasses should implement the methods to provide specific transaction management.

Example

>>> class UserUnitOfWork(UnitOfWork[User, UUID]):
...     def commit(self) -> None:
...         pass
...
...     def rollback(self) -> None:
...         pass
...
...     def repositories(self) -> dict[str, Repository[User, UUID]]:
...         pass

abstractmethod commit() → None

Commit all changes made within the unit of work.

abstractmethod repositories() → dict[str, Repository[T, ID]]

Get the repositories managed by this unit of work.

Returns:

A dictionary of repositories, keyed by name.

Return type:

dict[str, Repository[T, ID]]

abstractmethod rollback() → None

Roll back all changes made within the unit of work.

class flask_boilerplate.domain.primitives.ValueObject

Bases: ABC

Abstract base class for all value objects in the domain layer.

A value object is an immutable object that is defined by its attributes. It does not have a unique identity and is compared based on its attribute values.

Implements core value object characteristics: - Immutability - Structural equality - Validation - Serialization/deserialization

Subclasses should define their attributes using dataclass fields. The frozen=True parameter ensures immutability.

Inheriting classes must implement: - _validate() method - to_primitives() method

Example

>>> @dataclass(frozen=True)
>>> class Money(ValueObject):
...     amount: float
...     currency: str
...
>>> money1 = Money(amount=100.0, currency="USD")
>>> money2 = Money(amount=100.0, currency="USD")
>>> money1 == money2
True

abstractmethod __eq__(other: Any) → bool

Compare two value objects based on their attributes. :param other: The other value object to compare with. :type other: Any

Returns:

True if the value objects have the same attributes, False otherwise.

Return type:

bool

abstractmethod __hash__() → int

Generate a hash value for the value object based on its attributes. :returns: The hash value of the value object. :rtype: int

__post_init__() → None

Post-initialization hook to ensure the value object is properly validated.

validate() → None

Validate the value object’s data. :raises ValueObjectsError: If the value object’s data is invalid.