dddmisc-domain

`d3m.domain.entities`

`Entity`

Bases: IEntity, BaseModel, Generic[_ReferenceType]

Class Entity

A base class representing an entity.

Attributes:

Name	Type	Description
`__reference__`	`_ReferenceType`	The reference of the entity.

Examples:

Create an entity class:

>>> class Person(Entity):
...     name: str
...     surname: str
...
>>> person = Person(name="John", surname="Doe")
>>> assert isinstance(person.__reference__, UUID)

by default reference type is UUID and default factory is uuid.uuid4()

Set custom reference value:

>>> import uuid
>>> person = Person(name="John", surname="Doe", __reference__=uuid.UUID(int=1))
>>> assert person.__reference__ == uuid.UUID(int=1)

Set custom reference type:

>>> PersonId = int
>>> class Person(Entity[PersonId]):
...     name: str
...     surname: str
>>> # when set custom reference type, the default factory is not set
>>> person = Person(name="John", surname="Doe", __reference__=1)
>>> assert person.__reference__ == 1

Set default factory:

>>> import random
>>> class Person(Entity[PersonId], default_factory=random.randrange(1, 1000000)):
...     name: str
...     surname: str
>>> person = Person(name="John", surname="Doe")
>>> assert isinstance(person.__reference__, int)
>>> assert 1 <= person.__reference__ < 1000000

`reference: _ReferenceType` `property`

Returns the reference of the entity.

Returns:

Name	Type	Description
`_ReferenceType`	`_ReferenceType`	The reference of the entity.

`RootEntity`

Bases: Entity, IRootEntity, Generic[_ReferenceType]

RootEntity

A root entity is a base class for domain entities that serves as the root of an aggregate. It provides common functionality for managing events and versioning.

Attributes:

Name	Type	Description
`__reference__`	`_ReferenceType`	the reference of the root entity.
`__domain_name__`	`DomainName`	the domain name associated with the root entity
`__version__`	`Version`	The current version of the root entity.

Methods:

Name	Description
`create_event`	Create a domain event for this entity.
`collect_events`	Collect all the events that have been created for this entity.

Examples:

Set domain of root entity:

>>> class Person(RootEntity, domain='person'):
...     name: str
...     surname: str
>>> assert Person.__domain_name__ == DomainName('person')
>>> person = Person(name="John", surname="Doe")
>>> assert person.__domain_name__ == DomainName('person')
>>> assert isinstance(person.__reference__, UUID)
>>> assert person.__version__ == 1

Set custom version of root entity:

>>> person = Person(name="John", surname="Doe", __version__=2)
>>> assert person.__version__ == 2

Create and collect domain events for this entity:

>>> events = list(person.collect_events())
>>> assert len(events) == 0
>>> person.create_event('PersonCreated',
...                     reference=person.__reference__,
...                     name=person.name,
...                     surname=person.surname)
>>> events = list(person.collect_events())
>>> assert len(events) == 1
>>> event = events[0]
>>> assert isinstance(event, AbstractEvent)
>>> assert event.__domain_name__ == DomainName('person')
>>> assert event.__message_name__ == MessageName('PersonCreated')
>>> events = list(person.collect_events())
>>> assert len(events) == 0

`__domain_name__: DomainName` `property`

Get the domain name associated with the current class.

Returns:

Name	Type	Description
`DomainName`	`DomainName`	The domain name associated with the class.

`version: Version` `property`

Get the current version of the root entity.

Returns:

Name	Type	Description
`Version`	`Version`	The current version of the root entity.

`collect_events()`

Collects and returns events from this root entity.

Returns:

Type	Description
`Iterable[AbstractEvent]`	An iterable of `AbstractEvent` objects.

`create_event(__name, /, **payload)`

Create a domain event for this root entity.

Attributes:

Name	Type	Description
`__name`	`MessageName \| str`	The name of the event. Event class required be declared for domain of the root entity.
`**payload`	`MessageName \| str`	Additional keyword arguments to be passed as payload data for creating the event.

`increment_version(entity)`

Increments the version of the root entity

Attributes:

Name	Type	Description
`entity`	`RootEntity`	The root entity object whose version needs to be incremented.

`d3m.domain.bases`

`BaseDomainMessage`

Bases: BaseModel, IMessage, ABC

Class representing a base domain message.

Attributes:

Name	Type	Description
`__domain_name__`	`DomainName`	The domain name associated with the current class.
`__message_name__`	`MessageName`	The message name associated with the current class.
`__payload__`	`dict`	The payload of the message.
`__reference__`	`UUID`	The unique identifier for the message.
`__timestamp__`	`datetime`	The timestamp of when the message was created.

Methods:

Name	Description
`to_dict`	Returns a dictionary representation of the message.
`to_json`	Returns a JSON string representation of the message.

`d3m.domain.command`

`DomainCommand`

Bases: BaseDomainMessage, AbstractCommand

Class representing a base domain command.

Attributes:

Name	Type	Description
`__domain_name__`	`DomainName`	The domain name associated with the current command.
`__message_name__`	`MessageName`	The message name associated with the current command.
`__payload__`	`dict`	The payload of the command.
`__type__`	`MessageType`	Always return `MessageType.COMMAND`
`__reference__`	`UUID`	The unique identifier for the command.
`__timestamp__`	`datetime`	The timestamp of when the command was created.

Methods:

Name	Description
`to_dict`	Returns a dictionary representation payload of the command.
`to_json`	Returns a JSON string representation payload of the command.
`load`	class method returns new instance of the domain command

Examples:

Create command class:

>>> from d3m.domain import DomainCommand
>>> class CreatePerson(DomainCommand, domain='person'):
...     name: str
...     surname: str
>>> assert CreatePerson.__domain_name__ == DomainName('person')
>>> assert CreatePerson.__message_name__ == MessageName('CreatePerson')
>>> assert CreatePerson.__type__ == MessageType.COMMAND

Create command instance

>>> command = CreatePerson(
...               name='John',
...               surname='Black'
...           )
>>> assert command.__domain_name__ == DomainName('person')
>>> assert command.__message_name__ == MessageName('CreatePerson')
>>> assert isinstance(command.__reference__, UUID)
>>> assert isinstance(command.__timestamp__, datetime)
>>> assert command.__payload__ == dict(name='John', surname='Black')
>>> assert command.name == 'John'
>>> assert command.surname =='Black'

Load command from dict

>>> payload = dict(name='John', surname='Black')
>>> reference = uuid4()
>>> timestamp = datetime.now(timezone.utc)
>>> command = CreatePerson.load(payload, reference=reference, timestamp=timestamp)
>>> assert command.__domain_name__ == DomainName('person')
>>> assert command.__message_name__ == MessageName('CreatePerson')
>>> assert command.__reference__ == reference
>>> assert command.__timestamp__ == timestamp
>>> assert command.__payload__ == payload

Load command from json-string

>>> payload = '{"name":"John","surname":"Black"}'
>>> command = CreatePerson.load(payload, reference=uuid4(), timestamp=datetime.now())
>>> assert command.__domain_name__ == DomainName('person')
>>> assert command.__message_name__ == MessageName('CreatePerson')
>>> assert isinstance(command.__reference__, UUID)
>>> assert isinstance(command.__timestamp__, datetime)
>>> assert command.name == 'John'
>>> assert command.surname =='Black'

Command's payload to json serializeble dict

>>> assert command.to_dict() == {'name': 'John', 'surname': 'Black'}

Command's payload to json

>>> assert command.to_dict() == '{"name":"John","surname":"Black"}'

`get_command_class(domain, name)`

Return registered command class by domain and name

Attributes:

Name	Type	Description
`domain`	`DomainName \| str`	The domain of the event.
`name`	`MessageName \| str`	The name of the event.

Returns:

Type	Description
`AbstractCommandMeta`	The class of the command.

Examples:

>>> from d3m.domain import DomainCommand
>>> class CreatePerson(DomainCommand, domain='person')
...     name: str
...     surname: str
>>> command_class = get_command_class('person', 'CreatePerson')
>>> assert command_class is CreatePerson

`d3m.domain.event`

`DomainEvent`

Bases: BaseDomainMessage, AbstractEvent

Class representing a base domain event.

Attributes:

Name	Type	Description
`__domain_name__`	`DomainName`	The domain name associated with the current event.
`__message_name__`	`MessageName`	The message name associated with the current event.
`__payload__`	`dict`	The payload of the event.
`__type__`	`MessageType`	Always return `MessageType.EVENT`
`__reference__`	`UUID`	The unique identifier for the event.
`__timestamp__`	`datetime`	The timestamp of when the event was created.

Methods:

Name	Description
`to_dict`	Returns a dictionary representation payload of the event.
`to_json`	Returns a JSON string representation payload of the event.
`load`	class method returns new instance of the domain event

Examples:

Create event class:

>>> from d3m.domain import DomainEvent
>>> class PersonCreated(DomainEvent, domain='person'):
...     reference: UUID
...     name: str
...     surname: str
>>> assert PersonCreated.__domain_name__ == DomainName('person')
>>> assert PersonCreated.__message_name__ == MessageName('PersonCreated')
>>> assert PersonCreated.__type__ == MessageType.EVENT

Create event instance

>>> event = PersonCreated(
...             reference=person.__reference__
...             name=person.name,
...             surname=person.surname
...         )
>>> assert event.__domain_name__ == DomainName('person')
>>> assert event.__message_name__ == MessageName('PersonCreated')
>>> assert isinstance(event.__reference__, UUID)
>>> assert isinstance(event.__timestamp__, datetime)
>>> assert event.__payload__ == dict(reference=person.__reference__, name=person.name, surname=person.surname)
>>> assert event.reference == person.__reference__
>>> assert event.name == person.name
>>> assert event.surname == person.surname

Load event from dict

>>> payload = dict(
...               reference=person.__reference__
...               name=person.name,
...               surname=person.surname
...           )
>>> event = PersonCreated.load(payload, reference=uuid4(), timestamp=datetime.now())
>>> assert event.__domain_name__ == DomainName('person')
>>> assert event.__message_name__ == MessageName('PersonCreated')
>>> assert isinstance(event.__reference__, UUID)
>>> assert isinstance(event.__timestamp__, datetime)
>>> assert event.__payload__ == payload

Load event from json-string

>>> payload = '{"reference":"00000000-0000-0000-0000-000000000001","name":"John","surname":"Black"}'
>>> event = PersonCreated.load(payload, reference=uuid4(), timestamp=datetime.now())
>>> assert event.__domain_name__ == DomainName('person')
>>> assert event.__message_name__ == MessageName('PersonCreated')
>>> assert isinstance(event.__reference__, UUID)
>>> assert isinstance(event.__timestamp__, datetime)
>>> assert event.name == 'John'
>>> assert event.surname =='Black'
>>> assert event.reference == UUID(int=1)

Event's payload to json serializeble dict

>>> assert event.to_dict() == {'reference': '00000000-0000-0000-0000-000000000001', 'name': 'John', 'surname': 'Black'}

Event's payload to json

>>> assert event.to_dict() == '{"reference":"00000000-0000-0000-0000-000000000001","name":"John","surname":"Black"}'

`get_event_class(domain, name)`

Return registered event class by domain and name

Attributes:

Name	Type	Description
`domain`	`DomainName \| str`	The domain of the event.
`name`	`MessageName \| str`	The name of the event.

Returns:

Type	Description
`AbstractEventMeta`	The class of the event.

Examples:

>>> from d3m.domain import DomainEvent
>>> class PersonCreated(DomainEvent, domain='person')
...     name: str
...     surname: str
>>> event_class = get_event_class('person', 'PersonCreated')
>>> assert event_class is PersonCreated

`d3m.domain.exceptions`

`DomainError`

Bases: Exception

Custom Exception class for domain-specific errors

It allows for the definition of specific error messages and supports dynamic parameter substitution in the error message template.

Attributes:

Name	Type	Description
`__template__`	`str`
`__domain_name__`	`DomainName`	readonly attribute retrieves the domain name associated with the current class.
`__payload__`	`Mapping[str, Any]`	readonly attribute retrieves the payload (additional parameters) associated with the error.

Examples:

Creating a custom DomainError subclass

>>> class MyDomainError(DomainError, domain='my-domain'):
...     __template__ = "An error occurred in domain '{domain}'."
>>> assert MyDomainError.__domain_name__ == DomainName('my-domain')

Generating error message use template

>>> MyDomainError(domain='my-domain', foo='bar')
MyDomainError("An error occurred in domain 'test'.")

Get initialization error attribures

>>> MyDomainError(domain='my-domain', foo='bar').__payload__
{'domain': 'my-domain', 'foo': 'bar'}

Fail init error without template attributes

>>> MyDomainError(foo='bar')
TypeError: MyDomainError() missing 1 required keyword-only argument: 'domain'

Init with custom message, ignore template

>>> error = MyDomainError('custom message')
>>> error
MyDomainError('custom message')
>>> error.__payload__
{}

`get_error_class(domain, name)`

Return registered error class by domain and name

Attributes:

Name	Type	Description
`domain`	`DomainName \| str`	The domain of the error.
`name`	`str`	The name of the error.

Returns:

Type	Description
`type[DomainError]`	The class of the error.

Examples:

>>> from d3m.domain import DomainError
>>> class PersonNotFound(DomainError, domain='person')
...     __template__: str = 'Person {reference} not found'
>>> error_class = get_error_class('person', 'PersonNotFound')
>>> assert error_class is PersonNotFound

`get_or_create_error_class(domain, name, template=None)`

Get or create a error class associated with the given domain. If base error class associated with the given domain does not exist, then create it.

Attributes:

Name	Type	Description
`domain`	`str \| DomainName`	The domain of the error class.
`template`	`str \| None`	The template for the error class. Defaults to None.

Returns:

Type	Description
`type[DomainError]`	A type object representing the error class for the given domain and name.

Examples:

Get existing error class

>>> class BasePersonDomainError(DomainError, domain='person'):
...     pass
>>> class PersonNotFound(BasePersonDomainError):
...     __template__ = 'Person {reference} not found'
>>> error_class = get_or_create_error_class(
...     domain='person',
...     name='PersonNotFound',
... )
>>> assert error_class is PersonNotFound

Create new error class

>>> class BasePersonDomainError(DomainError, domain='person'):
...     pass
>>> error_class = get_or_create_error_class(
...     domain='person',
...     name='PersonNotFound',
...     template='Person {reference} not found'
... )
>>> assert error_class.__name__ == 'PersonNotFound'
>>> assert error_class.__domain_name__ = 'person'
>>> assert isinstance(error_class, BasePersonDomainError)

`get_or_create_base_error_class(domain, *, template=None)`

Get or create a base error class associated with the given domain

Attributes:

Name	Type	Description
`domain`	`str \| DomainName`	The domain of the base error class.
`template`	`str \| None`	The template for the base error class. Defaults to None.

Returns: The base error class associated with the given domain, or a newly created base error class if one does not exist.

Examples:

Get existing base error class

>>> class BasePersonDomainError(DomainError, domain='person'):
...     pass
>>> class PersonNotFound(BasePersonDomainError):
...     __template__ = 'Person {reference} not found'
>>> error_class = get_or_create_base_error_class('person')
>>> assert error_class is BasePersonDomainError

Create new base error class

>>> error_class = get_or_create_base_error_class('person')
>>> assert error_class.__module__ is __name__
>>> error_class.__name__
'__BaseError__'
>>> error_class.__domain_name__
DomainName('person')

dddmisc-domain

d3m.domain.entities

Entity

__reference__: _ReferenceType property

RootEntity

__domain_name__: DomainName property

__version__: Version property

collect_events()

create_event(__name, /, **payload)

increment_version(entity)

d3m.domain.bases

BaseDomainMessage

d3m.domain.command

DomainCommand

get_command_class(domain, name)

d3m.domain.event

DomainEvent

get_event_class(domain, name)

d3m.domain.exceptions

DomainError

get_error_class(domain, name)

get_or_create_error_class(domain, name, template=None)

get_or_create_base_error_class(domain, *, template=None)

`d3m.domain.entities`

`Entity`

`reference: _ReferenceType` `property`

`RootEntity`

`__domain_name__: DomainName` `property`

`version: Version` `property`

`collect_events()`

`create_event(__name, /, **payload)`

`increment_version(entity)`

`d3m.domain.bases`

`BaseDomainMessage`

`d3m.domain.command`

`DomainCommand`

`get_command_class(domain, name)`

`d3m.domain.event`

`DomainEvent`

`get_event_class(domain, name)`

`d3m.domain.exceptions`

`DomainError`

`get_error_class(domain, name)`

`get_or_create_error_class(domain, name, template=None)`

`get_or_create_base_error_class(domain, *, template=None)`