The introductory slides for a workshop on applying some classic patterns from Domain-Driven Design (DDD) in Python-based systems, which storage in a simple event store.
Is this the course for you? Domain-Driven Design (DDD) is an approach to software development that emphasizes high- fidelity modeling of the problem domain, and which uses a software implementation of the domain model as a foundation for system design. This approach helps organize and minimize the essential complexity of your software. Python is a highly productive, easy-to-learn, lightweight programming language that minimizes accidental complexity in the solution domain. This two day course will teach you the fundamentals of classical and modern DDD patterns in the context of Python. • We start by introducing the philosophy and principles of DDD. • We move quickly into hands-on domain modeling. • We implement a stand-alone domain model in pure Python code. • Taught on Windows, Linux or Mac OS X • Knowledge level of our Python for Programmers course is assumed Key Topics • Domain discovery • Bounded contexts and subdomains • Entity and value types • Aggregates • Domain events • Architecture • Persistence • Repositories • Projections • Domain services sixty-north.com
Design is... ‣ a philosophy of bringing software closer to the problem domain • involving domain experts ‣ a systematic approach intertwining design and development • strategic design practices • fundamental principles • guidelines, rules and patterns 7
with encapsulate with ENTITIES VALUE OBJECTS LAYERED ARCHITECTURE AGGREGATES REPOSITORIES act as root of FACTORIES encapsulate with express model with encapsulate with access with encapsulate with access with DOMAIN EVENTS express model with SERVICES push state change with CONTEXT MAP overlap allied contexts through minimize translation support multiple clients through formalize as CONTINUOUS INTEGRATION CUSTOMER /SUPPLIER CONFORMIST OPEN HOST SERVICE SEPARATE WAYS PUBLISHED LANGUAGE SHARED KERNEL relate allied contexts as free teams to go ANTI- CORRUPTION LAYER translate and insulate unilaterally with BOUNDED CONTEXT keep model unified by assess/overview relationships with UBIQUITOUS LANGUAGE names enter BIG BALL OF MUD segregate the conceptual messes define model within model gives structure to CORE DOMAIN work in autonomous, clean GENERIC SUBDOMAINS avoid overinvesting in cultivate rich model with express model with by Eric Evans http://domainlanguage.com/ddd/patterns/ Eric Evans
aggregate consistency • hosts commands (methods) which modify the aggregate • target for inbound aggregate references • all inter-aggregate references are by root entity ID
maintains aggregate consistency • hosts commands (methods) which modify the aggregate • target for inbound aggregate references • all inter-aggregate references are by root entity ID
entity (and thereby aggregate) construction • Allow us to express the ubiquitous language - verbs rather than nouns (constructors) • Hide construction details (e.g. entity ID generation)
Facilitate entity (and thereby aggregate) construction • Allow us to express the ubiquitous language - verbs rather than nouns (constructors) • Hide construction details (e.g. entity ID generation)
‣ Context maps • Integrations between bounded contexts • Logical mapping (concepts) • Physical mapping (actual software!) • Many integration patterns e.g. anti- corruption layer or open host service
with encapsulate with ENTITIES VALUE OBJECTS LAYERED ARCHITECTURE AGGREGATES REPOSITORIES act as root of FACTORIES encapsulate with express model with encapsulate with access with encapsulate with access with DOMAIN EVENTS express model with SERVICES push state change with CONTEXT MAP overlap allied contexts through minimize translation support multiple clients through formalize as CONTINUOUS INTEGRATION CUSTOMER /SUPPLIER CONFORMIST OPEN HOST SERVICE SEPARATE WAYS PUBLISHED LANGUAGE SHARED KERNEL relate allied contexts as free teams to go ANTI- CORRUPTION LAYER translate and insulate unilaterally with BOUNDED CONTEXT keep model unified by assess/overview relationships with UBIQUITOUS LANGUAGE names enter BIG BALL OF MUD segregate the conceptual messes define model within model gives structure to CORE DOMAIN work in autonomous, clean GENERIC SUBDOMAINS avoid overinvesting in cultivate rich model with express model with by Eric Evans http://domainlanguage.com/ddd/patterns/ Eric Evans
id: A unique identifier. version: An integer version. discarded: True if this entity should no longer be used, otherwise False. """ def __init__(self, id, version): self._id = id self._version = version self._discarded = False def _increment_version(self): self._version += 1 @property def id(self): """A string unique identifier for the entity.""" self._check_not_discarded() return self._id @property def version(self): """An integer version for the entity.""" self._check_not_discarded() return self._version @property def discarded(self): """True if this entity is marked as discarded, otherwise False.""" return self._discarded def _check_not_discarded(self): if self._discarded: raise DiscardedEntityError("Attempt to use {}".format(repr(self))) class DiscardedEntityError(Exception): """Raised when an attempt is made to use a discarded Entity.""" pass pass id as argument (don't create a new id here) maintain a discarded flag consider maintaining a version
id: A unique identifier. version: An integer version. discarded: True if this entity should no longer be used, otherwise False. """ def __init__(self, id, version): self._id = id self._version = version self._discarded = False def _increment_version(self): self._version += 1 @property def id(self): """A string unique identifier for the entity.""" self._check_not_discarded() return self._id @property def version(self): """An integer version for the entity.""" self._check_not_discarded() return self._version @property def discarded(self): """True if this entity is marked as discarded, otherwise False.""" return self._discarded def _check_not_discarded(self): if self._discarded: raise DiscardedEntityError("Attempt to use {}".format(repr(self))) class DiscardedEntityError(Exception): """Raised when an attempt is made to use a discarded Entity.""" pass pass id as argument (don't create a new id here) maintain a discarded flag consider maintaining a version
services. !"" bounded_context One package per bounded context # !"" __init__.py # $"" domain # !"" __init__.py # !"" model # # !"" __init__.py # # !"" aggregate_large Large complex aggregates should be their own package # # # !"" __init__.py This will expose the API of the aggregate at package level # # # !"" root_entity.py The root entity - responsible for consistency of the aggregate # # # !"" other_entity.py Another entity which forms part of this aggregate # # # !"" factories.py Factory functions for creating aggregates # # # $"" repository.py Aggregate persistence (Abstract if following hexagonal architecture) # # !"" aggregate_small.py Small, simple aggregates (most of them) can be a single file module # # $"" events.py Singleton publish-subscribe hub for domain events. Event base class. # $"" services # !"" __init__.py # !"" calculation_service.py Implement significant computations which involve multiple aggregates # $"" manipulation_service.py Perform significant manipulations involving for than one aggregate (care!) $"" infrastructure !"" __init__.py $"" repositories !"" __init__.py !"" large_aggregate_repo.py A concrete repository - in terms of the persistence infrastructure $"" small_aggregate_repo.py for each abstract repository type on the domain model Remember to use the ubiquitous language!
Args: event_predicate: A callable predicate which is used to identify the events to which to subscribe. subscriber: A unary callable function which handles the passed event. """ if event_predicate not in _event_handlers: _event_handlers[event_predicate] = set() _event_handlers[event_predicate].add(subscriber) def unsubscribe(event_predicate, subscriber): """Unsubscribe from events. Args: event_predicate: The callable predicate which was used to identify the events to which to subscribe. subscriber: The subscriber to disconnect. """ if event_predicate in _event_handlers: _event_handlers[event_predicate].discard(subscriber) def publish(event): """Send an event to all subscribers. Each subscriber will receive each event only once, even if it has been subscribed multiple times, possibly with different predicates. Args: event: The object to be tested against by all registered predicate functions and sent to all matching subscribers. """ matching_handlers = set() for event_predicate, handlers in _event_handlers.items(): if event_predicate(event): matching_handlers.update(handlers) for handler in matching_handlers: handler(event)
this domain. DomainEvents are value objects and all attributes are specified as keyword arguments at construction time. There is always a timestamp attribute which gives the event creation time in UTC, unless specified. Events are equality comparable. """ def __init__(self, timestamp=_now, **kwargs): self.__dict__['timestamp'] = utc_now() if timestamp is _now else timestamp self.__dict__.update(kwargs) def __setattr__(self, key, value): raise AttributeError("DomainEvent attributes are read-only") def __eq__(self, rhs): return self.__dict__ == rhs.__dict__ def __ne__(self, rhs): return self.__dict__ != rhs.__dict__ def __hash__(self): return hash(tuple(self.__dict__.items())) def __repr__(self): return self.__class__.__qualname__ + "(" + ', '.join( "{0}={1!r}".format(*item) for item in self.__dict__.items()) + ')'
def all_users(self, user_ids=None): return self.users_where(lambda user: True, user_ids) def users_with_name(self, name, user_ids=None): return self.users_where(lambda user: user.name == name, user_ids) def users_with_id(self, user_id): try: return exactly_one(self.all_users((user_id,))) except ValueError as e: raise ValueError("No WorkItem with id {}".format(user_id)) from e @abstractmethod def users_where(self, predicate, user_ids=None): raise NotImplementedError convenience queries for retrieving users implemented in terms of a generic query Repository will be subclassed with an infrastructure-specific implementation, which can specialise all queries. subclass implementation must override at least this
def all_users(self, user_ids=None): return self.users_where(lambda user: True, user_ids) def users_with_name(self, name, user_ids=None): return self.users_where(lambda user: user.name == name, user_ids) def users_with_id(self, user_id): try: return exactly_one(self.all_users((user_id,))) except ValueError as e: raise ValueError("No WorkItem with id {}".format(user_id)) from e @abstractmethod def users_where(self, predicate, user_ids=None): raise NotImplementedError convenience queries for retrieving users implemented in terms of a generic query Repository will be subclassed with an infrastructure-specific implementation, which can specialise all queries. subclass implementation must override at least this