---------------------------------------------------------------------- This is the API documentation for the hexkit library. ---------------------------------------------------------------------- ## Core Types & Base Classes Shared type aliases and provider base classes ## General Utilities Dependency-light helper functions and related errors ## Correlation IDs Correlation-ID context management and helpers ## OpenTelemetry Observability / tracing configuration ## Logging Structured logging configuration and formatters ## Protocols — Events Ports for event publishing and subscription ## Protocols — Data Access Objects Ports for DAOs and outbox publishing/subscription ## Protocols — Storage Ports for key-value and object storage ## Provider — Apache Kafka Event pub/sub backed by Apache Kafka ## Provider — MongoDB DAO and key-value stores backed by MongoDB ## MongoDB Migration Tooling Versioned database migration framework for MongoDB ## Provider — MongoDB + Kafka Persistent / outbox event publishing ## Provider — S3 Object storage and key-value stores backed by S3 ## Provider — Redis Key-value stores backed by Redis ## Provider — HashiCorp Vault Key-value stores backed by Vault ## Provider — In-Memory (Testing) In-memory providers and mocks for tests ---------------------------------------------------------------------- This is the User Guide documentation for the package. ---------------------------------------------------------------------- ## Overview ### User Guide This user guide covers the functional components of `hexkit` — providers, test utilities, and observability tools — alongside the architectural concepts that guide its development, with example code and guidance on when to use each feature. ## What's next? Familiarize yourself with the [library structure and conventions](structure.qmd), or explore the [architectural concepts](arch_concepts/index.qmd) that this library helps you implement. ## Other sections - [Protocols](protocols/index.qmd) — the ports that define hexkit's abstractions - [Providers](providers/index.qmd) — concrete backends implementing those protocols - [Observability Tools](observability_tools/index.qmd) — correlation IDs and logging - [Developer Help](developer_help/index.qmd) — practical guidance and test fixtures - [Glossary](glossary/index.qmd) — key terms used throughout the docs - [API Reference](../reference/index.qmd) — the full class and function reference ### Library Structure and Conventions TBD (explain folder structure and naming conventions) ## Architectural Concepts ### Architectural Concepts This section introduces some key architectural concepts within hexkit. The hexkit library facilitates the implementation of the Triple Hexagonal Architecture pattern. This pattern is an extension of the traditional Hexagonal Architecture (also known as Ports and Adapters pattern) that addresses code redundancy issues commonly found in microservice environments. To understand the Triple Hexagonal Architecture, it's important to first understand the traditional Hexagonal Architecture that it builds upon. Additionally, hexkit supports and integrates with other architectural patterns commonly used in modern microservice architectures, including event-driven communication and robust error handling mechanisms. ## Concepts - [Hexagonal Architecture](hexagonal_arch.qmd) - the foundational architecture pattern - [Triple Hexagonal Architecture](triple_hexagonal_arch.qmd) - hexkit's extension to address microservice code redundancy - [The DAO Pattern](dao_pattern.qmd) - Database access through Data Access Objects, hexkit's approach to persistence - [Event Driven Architecture](event_driven_arch.qmd) - Communication patterns supported by hexkit for decoupled microservices - [Dead Letter Queue](dlq.qmd) - Error handling and reliability patterns integrated with hexkit ### Hexagonal Architecture ## Introduction Hexagonal Architecture, also known as the "Ports and Adapters" pattern, is a software architecture pattern originally described by Alistair Cockburn in 2005. It aims to create loosely coupled application components that can be easily connected to their software environment through ports and adapters, while maintaining a clean separation between purely technical concerns and the core business logic. The pattern addresses fundamental challenges that became apparent when developers started following Domain-Driven Design (DDD) principles: how to keep the domain model pure and isolated from infrastructure concerns while maintaining testability and flexibility. ## The Core Concept The fundamental idea behind Hexagonal Architecture is to isolate the core business logic of an application from external concerns such as databases, user interfaces, web frameworks, and other external systems. This separation prevents technical concerns from bleeding into the domain logic through the introduction of an adapter layer. ### The Hexagonal Shape The architecture is called "hexagonal" because it's often depicted as a hexagon with the application core at the center, surrounded by ports and adapters. However, the exact shape doesn't matter—it could be any polygon or even a circle. The key principle is the clear separation between "the inside" (application core) and "the outside" (external systems) using an adapter layer. ![Separation of the application "inside" from the periphery](../img/hex_arch_1.jpg) ### Clean Architecture A similar software design was formalized by Robert C. Martin in 2012 under the name "Clean Architecture". It uses different terminology but emphasizes the same separation of concerns. ## Relationship to Domain-Driven Design Hexagonal Architecture nicely complements Domain-Driven Design (DDD) by introducing an "anti-corruption layer" between the application layer and technical infrastructure. The core idea of DDD was to focus on the domain model and keep it pure, isolated from infrastructure concerns. ![Hexagonal architecture and DDD](../img/hex_arch_2.jpg) This architectural approach allows major architectural decisions to be deferred, as the domain logic remains decoupled from specific technical implementations. ## Architecture Components ### Application Core ("The Inside") The center of the hexagon contains the business logic and domain models. This core: - Contains no references to external frameworks or technologies. - Uses only domain language and concepts. - Defines ports (interfaces) for external interactions. - Implements the core business rules and logic. - Remains focused on solving the specific domain/business problem. ### Ports Ports are interfaces that define the contract between the application core and the external world. They serve as the interaction points with the outside world using domain-specific language. Ports come in two types: #### Inbound Ports (Primary/Driving Ports) These are interfaces that allow external actors to interact with the application. They represent ways the outside world can "drive" the application by putting it out of an idle state and giving it tasks to work on. #### Outbound Ports (Secondary/Driven Ports) These are interfaces that the application uses to interact with external systems. They are "driven" by the application when it needs to perform operations outside its core domain. ### Adapters Adapters are concrete implementations of the ports that handle the translation between the domain language and the specific technology being used. They exist outside the core application codebase and connect the application ports with external infrastructure. #### Inbound Adapters (Primary/Driving Adapters) These adapt external requests to the application's inbound ports. #### Outbound Adapters (Secondary/Driven Adapters) These implement the outbound ports and handle communication with external systems. ### Examples of Ports and Adapters **Inbound (Primary/Driving) Examples:** - **Port**: User interface abstraction for handling user commands. - **Adapter**: REST API controller that translates HTTP requests to domain operations. - **Port**: Event processing interface for handling external events. - **Adapter**: Apache Kafka consumer that processes messages from a topic. - **Port**: Command-line interface for batch operations. - **Adapter**: CLI handler that processes command-line arguments and options. **Outbound (Secondary/Driven) Examples:** - **Port**: Repository interface for data persistence using domain language. - **Adapter**: PostgreSQL or MongoDB DAO implementation. - **Port**: External service integration interface. - **Adapter**: HTTP client for communicating with external APIs. - **Port**: Event publishing interface for domain events. - **Adapter**: Message queue publisher for Apache Kafka or RabbitMQ. The following diagram illustrates how these components interact, showing the application core at the center with various ports and adapters connecting it to external systems and actors. ![Schematic representation of the Hexagonal Architecture design pattern](../img/hex_arch_3.jpg) ## Benefits of Hexagonal Architecture Let's summarize the main benefits of this architecture again: ### Domain Focus Development can primarily focus on the logic needed to solve the domain problem, which is the actual motivation for the software project. The domain logic can be worked on in isolation from infrastructural details. ### Testability The separation of concerns makes it easy to test the business logic in isolation using true unit tests. Mock implementations of outbound ports can be provided for testing, while inbound adapters are not relevant for unit testing. This: - Improves testing quality by cleanly separating unit from integration tests. - Significantly increases test execution speed. - Enables effective test-driven development. ### Flexibility and Technology Independence Different adapters can be plugged in without changing the core business logic. The application becomes less coupled to specific environments or infrastructural setups. For example: - Switch from REST API to GraphQL API. - Change from MongoDB to PostgreSQL. - Replace message brokers without touching core logic. ### Maintainability The application core becomes easier to understand, which simplifies maintenance and helps onboard new developers. Domain complexity is not further inflated by technological complexity. ### Infrastructural Agnosticism Changes to infrastructure requirements often only affect the thin adapter layer, not the complex domain logic. This is particularly valuable for open-source projects, as it increases software reach by allowing easy integration into different infrastructure stacks. ### Evolutionary Architecture The pattern supports architectural evolution over time, allowing teams to explore different technical options in parallel or postpone technology decisions until more information is available. ## References - [Original proposal of the hexagonal architecture by Alistair Cockburn](https://alistair.cockburn.us/hexagonal-architecture/) - [Introduction to the hexagonal architecture based on code examples](https://www.youtube.com/watch?v=22WUhddwkS8) - [Presentation about "Clean Architecture" by Robert C. Martin](https://www.youtube.com/watch?v=Nsjsiz2A9mg) - [Domain-Driven Design: Tackling Complexity in the Heart of Software by Eric Evans](https://www.youtube.com/watch?v=dnUFEg68ESM) ### Triple Hexagonal Architecture ## Introduction Triple Hexagonal Architecture, the design pattern embodied by hexkit, is an extension of Hexagonal Architecture. If you are not already familiar with Hexagonal Architecture, please read the [Hexagonal Architecture](hexagonal_arch.qmd) chapter first. ## The Motivation The Hexagonal Architecture design pattern provides a clean separation between an application's core (i.e., the business logic or domain logic) and its outside dependencies such as databases, event/message brokers, or client delivery mechanisms (REST, RPC, WebSocket, etc.). As such, the pattern is principally applicable to both monoliths and microservices. However, a common issue in microservice settings is high code redundancy or code replication across microservice repositories. While this code redundancy is a general issue of microservices and not specific to Hexagonal Architecture, the design pattern on its own provides no solution to it. Fortunately, however, the redundancy in hexagonally designed microservices is highly concentrated in the adapter code, which is not specific to the domain use case(s) of a microservice but largely consists of cross-cutting logic to interface with external infrastructure. Here we present an extension to the Hexagonal Architecture design pattern that addresses the code redundancy by providing a systematic approach for separating service-specific from service-independent adapter logic. ## The Core Concept In the traditional Hexagonal Architecture, an adapter mediates between the application interface (a.k.a. "ports"), which is written in the language of the domain, and the infrastructure interface, which uses technology-specific jargon: ![Schematic representation of the traditional Hexagonal Architecture design pattern](../img/hex_arch_3.jpg) In the proposed extension, a third interface referred to as a protocol is added that splits the adapter into two parts as depicted in this diagram: ![Schematic representation of the Triple Hexagonal Architecture design pattern](../img/triple_hex_arch_1.jpg) A protocol describes an abstract concept of a type of infrastructure without exposing all the details of a specific implementation of that infrastructure. The only function of the inner part of an adapter is to translate between a port (of the application interface) and the protocol. Since both of these interfaces (port and protocol) are abstract and not bound to a specific type of infrastructure, we refer to this part of the adapter as the "abstract translator" or simply "translator". The second (outer) part of the adapter provides an implementation of the protocol that is infrastructure-specific. This second adapter part is referred to as a "protocol provider" or simply "provider". Since the protocol interface is depicted as a third, additional hexagon, we propose "Triple Hexagonal Architecture" as the name for this design pattern. ## The Protocol Definition Matters The protocol is the key ingredient of the Triple Hexagonal Architecture as it defines the separation between the two adapter fragments (protocol provider and abstract translator). Therefore, it is important to understand the intentions behind the protocol: The aim of a protocol definition is not necessarily to find an abstraction that most accurately reflects the underlying concept or mechanism of a specific type of infrastructure (such as event brokers). The primary aim of the protocol definition is to provide a pragmatic separation between, on the one hand, adapter logic that is service- or codebase-specific (i.e., the abstract translator), and, on the other hand, general-purpose adapter logic that can be shared across services or codebases (i.e., the protocol provider). ## The Perspective of the Application Interface Generally, the ports that together define the application interface should only use domain language and should avoid assumptions about the nature of the surrounding infrastructure. To more concretely outline what this means, we propose the following terms that may be used to classify and describe ports and associated operations: ### Emit An emit operation or emission is an outbound operation that publishes data to the outside world in a "fire-and-forget" fashion. From the perspective of the emitter, it remains unclear who is consuming that data and whether the data is correctly processed on the receiving side. ### Receive A receive operation is inbound and consumes data from an emitter without sending a response. An inbound receive operation in one application (receiver) corresponds to an outbound emit operation produced by another application (emitter). A natural fit for implementing emitter and receiver adapters is the use of event brokers such as Apache Kafka. However, in principle, the interaction could also be realized via protocols such as HTTP. ### Query A query operation is outbound; however, unlike emit operations, a response is mandatory. The response can contain an extensive payload or it could be a simple "OK". In both cases, the response should not only indicate that the data sent by the querier was received but also that it was successfully processed by the consumer. ### Serve A serve operation performed by one application (servant) is the inbound counterpart of a query operation performed by another application (querier). This interaction represents a point-to-point contact between the servant and the querier. A serve operation must always send back a response even if the response only confirms the successful processing of the received data. Moreover, multiple query-serve interactions may or may not happen as part of a transactional scope. A natural fit for implementing non-transactional serve and query adapters is the use of REST or gRPC. Database queries are an example that could be modeled as transactional query-serve relationships between an application and its database server. We recommend that a port should only be classified using one of the above terms and it should not make additional assumptions about the technical nature of the interaction and specifically the data transmission mechanism. This recommendation should outline the difference in intent between protocols and ports. A schematic representation of operation types from the perspective of the application - the perspective used to define the application interface (a.k.a. "ports"): ![Schematic representation of operation types from the perspective of the application](../img/triple_hex_arch_2.jpg) ## Operation Examples To further illustrate the Triple Hexagonal Architecture concept, below, we have listed examples for inbound and outbound operations. The examples are centered around protocols that correspond to infrastructure types that are commonly used in a microservice setting. The steps involved in the following examples correspond to this visual legend: ![Steps in inbound and outbound operations](../img/triple_hex_arch_3.jpg) ### Consuming Events from a Broker **Receive** operation: (a) Apache Kafka may serve as an event broker. (Potential alternatives: AWS Kinesis, or even RabbitMQ) (b) A protocol provider specific for consuming events from Apache Kafka may be used. Internally, it handles the communication with the broker, e.g., by using the kafka-python client library. (There may be alternative providers, e.g., for AWS Kinesis or RabbitMQ that may serve as a drop-in replacement. Moreover, mock or in-memory providers might be used for testing.) (c) This protocol describes the abstract concept of consuming an event from an event broker. It is aware of the following notions: event type (the type of event, e.g., "user_account_upserted" or "user_account_deleted"), event key (all events with the same key are guaranteed to be delivered in order), event payload (the actual content/data that is shipped with the event), event schema (the shape of the payload which is defined by the event type), and topics (high-level organization of events into logs). Technical details on the underlying implementation are not specified by the protocol. How this vocabulary is used may vary considerably between protocol providers. Moreover, providers may choose to ignore some of the vocabularies if it is not relevant for the corresponding technology (e.g., a RabbitMQ provider might ignore the "event key" term since AMQP has no concept compatible with the notion of an event key). (d) The abstract translator is able to translate the vocabulary used by the protocol into domain language used by the port. Thus, it knows how to convert the schema of an incoming event with a specific type (e.g., "user_account_upserted") into a domain representation (e.g., a domain entity object according to Eric Evans' DDD). (e) The port is not aware of how events are organized into topics or how the payload was formatted. It might not even know (or need to know) that the data was passed in as an event. From the perspective of the port, this operation is simply seen as a *receive operation* (as defined above) with no further understanding of the involved delivery mechanism or technology. Data is expected not as JSON schema but as domain objects. ### Publishing Events to a Broker **Emit** operation: (α) Similar to the above example, the port involved in event publishing is not aware of event delivery details. It sees the operation as a simple *emission* (as defined above) that sends out data in the form of domain objects. (β) The abstract translator understands the domain objects and it knows how to package and organize the corresponding event: it infers the event type and event key and decides to which destination topic this event belongs. Moreover, it brings the data into a shape compliant with the schema contract associated with the event type. The result is handed to the protocol (and the used provider). (γ) The protocol conveys an abstract understanding of how to publish events to an event broker. The used vocabulary is similar to the "event subscription" protocol: event type, event key, event payload, event schema, and topic. Again, the protocol is entirely independent of a specific broker technology and has no opinion on how to use the proposed vocabulary. (δ) The protocol provider may be a specific implementation for publishing events to Apache Kafka. Internally, it manages the communication with the broker, e.g., by using the `kafka-python` client library. (Alternative providers may exist for AWS Kinesis or RabbitMQ. Moreover, mock or in-memory versions may be used for testing.) (ε) Apache Kafka may serve as an event broker. (Potential alternatives: AWS Kinesis, or even RabbitMQ) ### Exposing a RESTful API **Serve** operation: (a) In this case, the outside interaction partner is a RESTful client which could be another service or a user with its browser running a web app. (b) The protocol provider is running the web server. Moreover, it might use a web framework such as FastAPI or Flask. There may be multiple providers based on different frameworks that can be used interchangeably: e.g., one ASGI provider based on FastAPI and another WSGI provider based on Flask. (c) The protocol itself is entirely framework agnostic. It only conveys an abstract understanding of resource-oriented RESTful operations whereby HTTP verbs are used to represent the action performed on a resource. (d) The abstract translator maps incoming REST requests obtained via the protocol interface to application-internal functions. The result of that function call is translated into an HTTP-compatible response format and handed back to the protocol. (e) The corresponding port and the underlying application logic are completely unaware of REST principles. From their perspective, the communication mechanism could very well be using gRPC instead of REST. The port would not even change if a non-web delivery mechanism such as a standard desktop GUI or a CLI would be used. The only information that the port has on the delivery mode is that it exhibits the characteristics of a *serve operation* (as defined above). The port is exposing interfaces to functions that correspond to the use cases of the application without any further assumptions on the applied technologies. ### Performing a RESTful Call to an External Service **Query** operation (non-transactional): (α) As for the above example, the port is entirely unaware of the nature of the used communication mechanism except that it exhibits the characteristics of a non-transactional *query operation* (as defined above). The port is formulating a request using domain language. (β) The abstract translator translates the domain request into a RESTful request that uses the vocabulary defined by the protocol. Thereby, it also adds the connection string that identifies the targeted API server. (This connection string is, however, typically further resolved by a service registry, an API gateway, or a reverse proxy.) (γ) The protocol describes the abstract principles of performing a resource-oriented RESTful call whereby HTTP verbs are used to represent the requested action. (δ) The protocol provider provides an implementation of the protocol interface. To do so, it might, e.g., use the `urllib.request` module from Python's standard library. (ε) The interaction partner who is receiving the query is an API server belonging to another service. ### Querying a Database **Query** operation (transactional): (α) Again, the port has no detailed understanding of the used data transfer mechanism. It only knows that it conforms to the characteristics of a (transactional) query operation. Moreover, the port makes the assumption that the sent data is persisted on the receiving side. Any other technical details on the data transfer or persistency mechanism are not relevant to the port. In principle, the port could be wired to a relational database like PostgreSQL, a NoSQL database like MongoDB, a graph database like ArangoDB, an event store based on Apache Kafka, or it could talk to an external RESTful service that internally manages a database of some kind. The port might be modeled using the repository pattern (according to Eric Evans' DDD) or the DAO pattern. Thereby, data is sent out and received as domain objects. (β) This part of the adapter translates between domain objects obtained or expected by the port and a simple data representation such as JSON that is handed to or obtained from the protocol. Moreover, it generates an abstract query statement that uses the vocabulary of the protocol. (γ) The protocol defines an abstract language for expressing basic CRUD database queries that are independent of a specific database implementation. (This simple CRUD logic might have no built-in understanding of relations between database entries as found in SQL databases.*) (δ) There may be a protocol provider that is specific for MongoDB. It converts the abstract query statement obtained via the protocol into a MongoDB-specific query, e.g., by using the Python library `pymongo`. (An alternative provider may be based on SQL databases and frameworks like SQLAlchemy. Moreover, one could provide an implementation that stores data as an event log in Apache Kafka.) (ε) The infrastructure might be powered by MongoDB. (Or alternatively realized via an SQL database or Apache Kafka-based event stores.) *Please note: here we provide a protocol example that does not make use of relations between database entries as found in SQL databases. To make use of this and more advanced features of SQL databases, an adapter designed according to the original, non-triple hexagonal architecture design pattern might make more sense as discussed in the following section.* ## Limitations and Exceptions Every design pattern has its use cases but also its limitations. The Triple Hexagonal Architecture is not different in that regard. However, it has the nice property that it is "non-greedy" in nature, i.e., it does not force you to fully commit to the pattern for your entire application. In fact, it is perfectly fine if you decide in a per-adapter/per-port manner whether the Triple Hexagonal Architecture pattern is applicable or whether a different design is more useful. Consider a port for database interactions as an example. To fully and conveniently utilize the relational nature of SQL databases, ORM frameworks such as SQLAlchemy are a common choice. However, designing a protocol that reflects all the required features of such an ORM framework into an abstract representation might be challenging. Thus, you might decide to fall back to the ordinary non-triple Hexagonal Architecture design in the case of this specific port. Technically, you can consider ORM frameworks like SQLAlchemy to be themselves already implementations of the Triple Hexagonal design. The part of the framework that handles ORM classes can be seen as the abstract translator. The SQL language can be seen as a generic protocol for addressing relational databases. The SQLAlchemy plugins that support different SQL dialects can be seen as different providers for the SQL protocol. So with SQLAlchemy you already get most of the benefits of Triple Hexagonal Design: you can switch between database implementations without changes to the ORM models (as long as you are committed to only using SQL-based databases and you are not using more exotic SQLAlchemy features not supported by all database implementations) and you can use simple mock or in-memory providers (in-memory SQLite) for testing. ### The DAO Pattern ## Introduction A Data Access Object (DAO) is an object that gives the rest of an application a small, well-defined interface for storing and retrieving data, while hiding which database is used and how the data is mapped to it. The pattern was popularized in the early 2000s by Sun's *Core J2EE Patterns* catalog, but the underlying idea is much older and not tied to any language or framework: business logic should say *what* it wants to persist or look up, and a dedicated object should worry about *how* that happens. In hexkit, the DAO pattern is one of the supported architecture patterns and the library's answer to database access in general. This chapter explains the pattern itself, why it is worth the extra indirection, and how it maps onto hexkit's [Triple Hexagonal Architecture](triple_hexagonal_arch.qmd). If you are looking for the concrete interfaces and code, jump ahead to the [DAO protocol](../protocols/dao.qmd) chapter or the [MongoDB DAO provider](../providers/mongodb.qmd#mongodb-dao). ## The Problem It Solves Consider a service that manages user accounts and talks to its database directly from the business logic: ```python async def deactivate_user(user_id: str): result = await db.users.update_one( {"_id": user_id}, {"$set": {"active": False}} ) if result.matched_count == 0: raise UserNotFound(user_id) ``` This works, but the domain logic (deactivating a user) is now entangled with MongoDB specifics: the `_id` convention, the `$set` update syntax, and the shape of pymongo's result objects. This entanglement has consequences that grow with the codebase: - **Testing gets heavy.** Every test of the business rule needs a real or carefully mocked database, even though the rule itself has nothing to do with MongoDB. - **The technology choice hardens.** Query syntax spreads through the codebase, so moving to a different database (or even a new major version of the driver) means touching business logic everywhere. - **Persistence concerns leak.** Details like retry behavior, index management, and error translation end up duplicated wherever the database is touched. The DAO pattern addresses this by pulling all persistence access behind an interface that speaks in terms of the application's data, not the database's: ```python async def deactivate_user(user_id: UUID): user = await user_dao.get_by_id(user_id) await user_dao.update(user.model_copy(update={"active": False})) ``` The business logic no longer knows or cares what is behind `user_dao`. In production it is MongoDB; in a unit test it can be a simple in-memory object. ## The Shape of the Pattern A classic DAO setup has three participants: 1. **A data transfer object (DTO)** describing the shape of the data being stored. In hexkit, DTOs are [Pydantic](https://docs.pydantic.dev) models, which gives you validation and serialization for free. 2. **An abstract DAO interface** declaring the available operations. Hexkit defines this once, generically, as the [`Dao`](../../reference/protocols.dao.Dao.qmd) protocol: CRUD operations (create, read, update, delete) plus `find` operations for querying by field values. 3. **A concrete implementation** of that interface for a specific storage technology, such as hexkit's [`MongoDbDao`](../../reference/providers.mongodb.MongoDbDao.qmd). Because the interface is generic over the DTO type, one DAO instance manages exactly one kind of resource (users, books, upload jobs, and so on), roughly corresponding to one table or collection in the database. A service typically holds one DAO per resource type it persists. ## DAO or Repository? You may have encountered the closely related *repository* pattern from Domain-Driven Design. The two are often confused, and in simple applications they can look identical. The difference is one of intent: - A **DAO** is persistence-oriented. It mirrors the storage structure (one DAO per table or collection) and offers generic data operations: insert this record, fetch that one, find all records matching these values. - A **repository** is domain-oriented. It presents itself as an in-memory collection of domain aggregates and speaks pure domain language, for example `overdue_loans_for(member)`. Hexkit deliberately provides the DAO as the generic, reusable building block, since a repository's domain-specific interface cannot be provided by a library. If your service calls for repositories, the natural approach in hexkit is to define the repository as a port of your application and implement it as a thin translator on top of one or more DAOs. For straightforward services, using the DAOs directly as outbound dependencies of your domain logic is perfectly fine. ## DAOs in the Triple Hexagonal Architecture Hexkit structures all infrastructure access according to the [Triple Hexagonal Architecture](triple_hexagonal_arch.qmd), which splits each adapter into a service-independent *provider* and a service-specific *translator*, connected through an abstract *protocol*. Database access via DAOs follows exactly this scheme: - **The port** is defined by your application in domain terms. From the application's perspective, persisting and retrieving data is a *transactional query operation*: it sends or requests data and relies on the answer, without assumptions about the storage technology. The port might be a repository interface, or it might simply be the DAO interface typed to your DTO model. - **The protocol** is hexkit's [`Dao`](../../reference/protocols.dao.Dao.qmd) interface together with the [`DaoFactoryProtocol`](../../reference/protocols.dao.DaoFactoryProtocol.qmd). It defines an abstract vocabulary for CRUD-style database access (resources, IDs, DTO models, find mappings, indexes) without committing to any particular database. The [DAO protocol](../protocols/dao.qmd) chapter describes it in detail. - **The provider** implements the protocol for a concrete technology. Hexkit currently ships a [MongoDB provider](../providers/mongodb.qmd#mongodb-dao) for production use and an [in-memory mock DAO](../providers/testing.qmd) for unit testing. Providers for other databases can be added without touching any service code, and the [DAO protocol](../protocols/dao.qmd#implementing-a-custom-provider) chapter shows how to write your own. - **The translator** is the small piece you write per service: it wires a DAO produced by the factory to your port, converting between domain objects and DTOs where the two differ. The [Querying a Database](triple_hexagonal_arch.qmd#querying-a-database) example in the [Triple Hexagonal Architecture](triple_hexagonal_arch.qmd) chapter walks through this exact scenario step by step. In practice, using a DAO looks like this: ```python from pydantic import BaseModel class Book(BaseModel): """A book in our catalog (the DTO model).""" id: str title: str author: str # The factory is provided by a concrete provider, e.g. MongoDbDaoFactory. # One DAO manages one resource type: book_dao = await dao_factory.get_dao(name="books", dto_model=Book, id_field="id") await book_dao.insert(Book(id="9780134757599", title="Refactoring", author="Fowler")) book = await book_dao.get_by_id("9780134757599") async for hit in book_dao.find_all(mapping={"author": "Fowler"}): print(hit.title) ``` Nothing in this code mentions MongoDB. Swapping the factory for another provider, for example the in-memory one during testing, changes the storage backend without changing a single line of the logic. ## When to Use a DAO, and When Not To The DAO protocol is intentionally simple. It models resources as self-contained documents identified by an ID, and supports CRUD operations plus filtering by field values. This makes it a great fit when: - your resources are document-shaped, with no complex relations between them, - CRUD operations and simple filtered lookups cover your query needs, - you want to unit test business logic without a running database, and - you value the freedom to change the storage backend later. It is deliberately *not* a general-purpose database abstraction. If your service leans heavily on relational features such as joins across many tables, rich aggregations, or complex multi-statement transactions, an ORM like SQLAlchemy or direct use of a database client will serve you better than forcing those needs through a CRUD-shaped interface. Crucially, this is not an all-or-nothing decision. The Triple Hexagonal Architecture is non-greedy: you decide per port whether the pattern applies. A service can happily use hexkit DAOs for its document-shaped resources and SQLAlchemy for a reporting feature next door. The [Limitations and Exceptions](triple_hexagonal_arch.qmd#limitations-and-exceptions) section of the Triple Hexagonal Architecture chapter discusses this trade-off in more depth, including why ORMs can themselves be viewed as an instance of the same design idea. ## Beyond Plain Persistence Because all database access flows through one narrow interface, hexkit can layer additional behavior on top of it. The most notable example is the *outbox pattern*: the [`DaoPublisher`](../protocols/daopublisher.qmd) variant of the DAO automatically publishes an event whenever a resource changes, so other services can follow the state of your resources without you writing any publishing code. See the [Publisher Variants](../developer_help/outbox_v_persistent_pub.qmd) chapter for when this is useful. ## Further Reading - [DAO protocol](../protocols/dao.qmd): the interfaces, operations, and error model in detail, including how to implement your own provider - [MongoDB DAO provider](../providers/mongodb.qmd#mongodb-dao): configuration, queries, indexes, and error handling with the production MongoDB implementation - [Triple Hexagonal Architecture](triple_hexagonal_arch.qmd): the architectural context that protocols and providers live in - [Core J2EE Patterns: Data Access Object](https://www.oracle.com/java/technologies/dataaccessobject.html): the original catalog entry that popularized the pattern ### Event-Driven Architecture This document provides an overview of the asynchronous communication methods between microservices. It offers both a high-level explanation of how information is sent and received and an examination of the layers of abstraction implemented within hexkit. The document also outlines how the outbox pattern is employed to solve certain challenges. In the following, we assume that Apache Kafka is used as the event streaming infrastructure, since this is the primary and currently only infrastructure that hexkit supports. However, the idea behind hexkit is that it could be extended to also support other event streaming infrastructure such as Apache Pulsar or, in principle, message brokers or event bus services. ## Event-driven Architecture with Apache Kafka > See the [Kafka documentation](https://kafka.apache.org/intro) for in-depth information on Kafka's inner workings. Event-driven architecture refers to a communication pattern where information is exchanged indirectly and often asynchronously by interacting with a broker. Rather than making direct API calls from one service to another, one service, called the *producer*, publishes information to a broker in the form of an *event* (also called a *message*), and is unaware of all downstream processing of the event. It is essentially "fire and forget". Other services, called *consumers*, can consume the event once it has been published. Similarly, just as the producer is unaware of an event's fate, consumers are unaware of its origin. This has important implications for design. In Kafka, events have metadata headers, a timestamp, a *key*, and a *payload*. To further distinguish events, hexkit adds a *type* field and an optional *event ID* field in the metadata headers. Events are organized into configurable *topics*, and further organized within each topic by key. A topic can have any number of producers and consumers, and messages are not deleted after consumption (enabling rereads). In the context of microservices, event-driven architecture provides some important benefits: 1. **Decoupling of Service Dependencies:** Services communicate by emitting events, rather than direct requests, reducing service coupling. Keeping contractual definitions (i.e. payload schemas) in a dedicated library means services can evolve with minimal change propagation. 2. **Asynchronous Communication:** Kafka facilitates asynchronous data processing, allowing services to operate independently without waiting for responses. This improves performance and fault tolerance, as services can continue functioning even if one component is slow or temporarily unavailable. 3. **Consistency and Order Guarantee:** Kafka maintains the order of events within a partition for a given key, crucial for consistency in processing events that are related or dependent upon one another. However, event-driven architecture also introduces challenges: 1. **Duplicate processing:** Because event consumers have no way to know whether an event is a duplicate or a genuinely distinct event with the same payload, consumers must be designed to be idempotent. 2. **Data Duplication:** If service A needs to access the data maintained by service B, the options are essentially an API call (which introduces security, performance and coupling concerns) or duplicating the data of service B for service A. Here, hexkit supports mitigating the issue via data duplication by way of the outbox pattern, described later in this document. 3. **Tracing:** A single request flow can span many services with a multitude of messages being generated along the way. The agnostic nature of producers and consumers makes it harder to tag requests than with traditional API calls, and several consumers can process a given message in parallel, sometimes leading to complex request flows. Here, hexkit enables basic tracing by generating and propagating a *correlation ID* (also called a request ID) as an event header. ## Abstraction Layers To simplify Kafka usage in services, hexkit features protocols and providers. Rather than implementing this functionality from scratch in every new project, hexkit provides the [`KafkaEventPublisher`](../providers/kafka/publisher.qmd) and [`KafkaEventSubscriber`](../providers/kafka/subscriber.qmd) provider classes, along with a [`KafkaConfig`](../providers/kafka/index.qmd#configuration) Pydantic configuration class. ### Producers When a service publishes an event, it uses the `KafkaEventPublisher` provider class along with a service-defined _translator_ (a class implementing the `EventPublisherProtocol`). The purpose of the translator is to provide a domain-bound API to the core of the service for publishing an event without exposing any of the specific provider-related details. The core calls a method like `publish_user_created()` on the translator, and the translator provides the requisite payload, topic, type, and key to the `KafkaEventPublisher`. The translator's module usually features a config class for defining the topics and event types used. It's important to note that a translator is not strictly required in order to use the `KafkaEventPublisher`. As long as the required configuration is provided, there's nothing to stop the core from using it directly. However, it will make switching to a different infrastructure (that might be supported by hexkit in the future) more difficult. ### Consumers While the `KafkaEventPublisher` can be used directly, the `KafkaEventSubscriber` always requires a *translator* to deliver event payloads to the service's core components. The translator is defined outside of hexkit (in the service code, not in hexkit itself) and implements the `EventSubscriberProtocol`. The provider passes the event to the translator through the protocol's `consume` method, which then calls an abstract method implemented by the translator. The translator receives the event and examines its attributes (the topic, payload, key, headers, etc.) to determine processing. Naturally, the actual processing logic varies, but usually the payload is validated against the expected schema and used to call a method in the service's core. When control is returned to the `KafkaEventSubscriber`, the underlying consumer (from `AIOKafka`) saves its offsets, declaring that it has successfully handled the event and is ready for the next. The [Kafka Event Subscriber](../providers/kafka/subscriber.qmd) chapter covers the translator pattern and offset handling in detail. A diagram illustrating the process of event publishing and consuming, starting with the microservice: ![Kafka abstraction](../img/kafka_basics_generic.png) ## Outbox Pattern Sometimes services need to access data owned by another service. The challenge of enabling this kind of linkage involves ensuring data consistency and avoiding coupling. The outbox pattern provides a way to mitigate these risks by publishing persisted data as Kafka events for other services to consume as needed. Some advantages include: - Events can be republished from the database as needed (fault tolerance) - The coupling between services is minimized - The data producer remains the single source of truth - Kafka events no longer need to be backed up separately from the database Hexkit's interfaces for the pattern are the [DAO Publisher](../protocols/daopublisher.qmd) protocol on the producing side and the [DAO Subscriber](../protocols/daosubscriber.qmd) protocol on the consuming side; each has its own chapter. ### Topic Compaction & Idempotence Topic compaction means retaining only the latest event per key in a topic. Why do this? In hexkit's implementation of the outbox pattern, published events are categorized as one of two event types: `upserted` or `deleted`. The entire state of the object, insofar as it is exposed by the producer, is represented in every event. Consumers interpret the event type and act accordingly, based on their needs (although this typically translates to at least upserting or deleting the event in their own database). When combined with topic compaction, the result is that consumers only have to get the latest event to be up to date with the originating service. ### Implications One requirement of using the outbox pattern as implemented in hexkit is that consumers must be idempotent. The event state is stored in the database and can be republished as needed, so services must be prepared to consume the same event multiple times. Event republishing can occur at any point, which means that sequence-dependent events might be re-consumed out of order. Depending on the complexity of the microservice landscape, it can be difficult to foresee all cases where out-of-order events could cause problems. This is why it is essential to perform robust testing and use features like topic compaction where necessary. ## Dead Letter Queue Event consumers are sometimes unable to process received events due to validation errors, data inconsistencies, or temporary service failures. Since service consumers are expected to be idempotent and long-lived, individual event processing failures cannot be allowed to crash the service. To address this challenge, hexkit provides a Dead Letter Queue (DLQ) mechanism, which automatically handles failed events. When an event consumer encounters an exception while processing an event, the `KafkaEventSubscriber` can be configured to: 1. Retry the event processing a configurable number of times 2. If retries are exhausted, publish the failed event to a dedicated DLQ topic 3. Continue processing other events instead of crashing Events in the DLQ retain their original payload, type, and key but include additional metadata headers with failure information (service name, original topic, exception details, etc.). This enables manual review and corrective action. Failed events can then be republished to a service-specific retry topic, where they are reprocessed with the original topic name restored. For a comprehensive understanding of the DLQ mechanism, configuration options, and recovery procedures, see the [Dead Letter Queue](dlq.qmd) documentation. ## Tracing Request flows are traced with a `correlation ID`, or request ID, which is generated at the request origin. The correlation ID is propagated through subsequent services by adding it to the metadata headers of Kafka events. When a service consumes an event, the provider automatically extracts the correlation ID and sets it as a [context variable](https://docs.python.org/3.9/library/contextvars.html#context-variables) before calling the translator's `consume()` method. In the case of the outbox mechanism, the correlation ID is stored in the document metadata. When initial publishing occurs, the correlation ID is already set in the context. For republishing, however, the correlation ID is extracted and set before publishing the event, thereby maintaining traceability for disjointed request flows. ### Dead Letter Queue (DLQ) ## Overview This document describes the Dead Letter Queue (DLQ) mechanism in hexkit, which provides robust error handling for event processing failures in Kafka-based event systems. The DLQ support allows services to gracefully handle failed event processing by redirecting problematic events to a dedicated queue for later analysis or reprocessing, rather than crashing the service. For information about general event subscription, see [Event Driven Architecture](./event_driven_arch.qmd). ## DLQ Lifecycle Apache Kafka doesn't provide out-of-the-box DLQ functionality, so hexkit bridges that gap. The Kafka event subscriber provider in hexkit can be configured to automatically handle a DLQ event flow with only minor changes in the standard usage. When an event consumer encounters an exception while processing an event, the system can: 1. Retry the event processing (if configured) 2. When retries are exhausted, publish the failed event to a dedicated DLQ topic 3. Continue processing other events instead of crashing ![DLQ Lifecycle Diagram](../img/numbered_dlq_flow.png) The flow diagram above demonstrates the general DLQ lifecycle: 1. Failed events are retried a configurable number of times. 2. Upon final failure, they are published to the configured DLQ topic. 3. Events in the DLQ are manually reviewed through an external solution. 4. DLQ events are eventually retried by publishing them to a service-specific `retry-*` topic. 5. Upon consuming an event from the retry topic, the consumer restores the original topic name and proceeds with the normal request flow. 6. The event is consumed again, this time successfully. - If the event fails again for some reason, the DLQ process restarts. ## DLQ Event Structure When an event is published to the DLQ, it maintains the original payload, type, and key, but includes additional headers with information about the failure: | Header | Description | | -------------- | ------------------------------------------------------- | | event_id | UUID4 unique to the event instance | | service | The name of the microservice where failure occurred | | original_topic | The topic where the event was originally published | | exc_class | The class name of the exception that caused the failure | | exc_msg | The error message from the exception | #### Example DLQ Event Assumes the following event is consumed from a `users` topic and results in an error: ```json { "payload": {"user_id": "abc123"}, "key": "abc123", "headers": { "type_": "user_registered", "correlation_id": "a648c68c-f14b-4d0a-8fc8-31824987613c", "event_id": "290de3f3-1a99-4c61-b71a-4074b65a4918", } } ``` That event then becomes the following when published to the DLQ topic: ```json { "payload": , "key": , "headers": { "type_": , "correlation_id": , "event_id": , "service": "my-service", "original_topic": "users", "exc_class": "ValueError", "exc_msg": "Invalid data format" } } ``` You can see that the DLQ event contains extra header information that can be used to deal with the event later. ## Consuming From the DLQ While hexkit makes it easy to both divert problematic events to a DLQ topic and reintroduce them once resolved, it does not provide a comprehensive toolbox for DLQ event resolution. The user is entirely responsible for monitoring and maintaining the DLQ topic and its events, although hexkit does provide some facilitating classes. GHGA, for example, uses a dedicated DLQ Service. For consuming events from the DLQ, hexkit provides the `DLQSubscriberProtocol`. This protocol extends the standard EventSubscriberProtocol with additional parameters for accessing the DLQ metadata. ## Republishing and Re-consuming DLQ Events Events that are dealt with in the DLQ topic and destined to be republished should receive a fresh `event_id` to mark them as distinct from the original. Furthermore, events should not be republished to their original topic. Instead, they should be republished to a special retry topic with a name in the format `retry-`. The service name should match the value configured for `KafkaConfig.service_name` and thus the value included in the DLQ supplementary header, `service`. For instance, if the service name is `abc`, then the reviewed event should be republished to a topic called `retry-abc`. The `abc` service will consume this event and the `KafkaEventSubscriber` class from hexkit will automatically substitute `retry-abc` with the the value for `original_topic` before passing the event to the service's event subscriber translator. The result is that the reintroduction procedure is completely transparent to the service's event subscriber translator, and the event gets processed identically to every other event. Checklist: - Generate a new UUID4 for the `event_id` header - Preserve the `original_topic` header - Do not include the `service`, `exc_msg`, or `exc_class` headers - Republish the event to `retry-` ## Configuration Parameters The DLQ functionality is controlled by several configuration parameters in the KafkaConfig class: | Parameter | Description | Default | Example | | --------------------- | -------------------------------------------------------------- | ------- | -------- | | `kafka_max_retries` | Maximum number of times to retry failed events | 0 | 3 | | `kafka_enable_dlq` | Toggle to enable/disable DLQ functionality | False | True | | `kafka_dlq_topic` | Topic name for the Dead Letter Queue | "dlq" | "my-dlq" | | `kafka_retry_backoff` | Base seconds to wait before retrying (doubles with each retry) | 0 | 2 | #### Minimal Example Configuration: ```python config = KafkaConfig( service_name="my-service", service_instance_id="instance-1", kafka_servers=["kafka:9092"], kafka_max_retries=3, kafka_enable_dlq=True, kafka_dlq_topic="dlq", kafka_retry_backoff=2 ) ``` ## Retry Mechanism The retry mechanism operates alongside and independently of the DLQ feature, but the two are intended to be used in concert. The DLQ filters events that would otherwise cause a service to crash, but oftentimes those failures are due to transient issues like database connection interruptions. The retry mechanism can help prevent clogging the DLQ with transient failures that don't represent genuine errors. The basic retry logic is straightforward: 1. If retries are enabled (`kafka_max_retries` > 0), the event is retried immediately 2. Each retry attempt uses exponential backoff based on the `kafka_retry_backoff` setting 3. The backoff time doubles with each retry attempt: `backoff_time = retry_backoff * 2^(retry_number - 1)` 4. If retries are exhausted and an error still occurs, the behavior depends on whether the DLQ is enabled. If the DLQ is enabled, the error bubbles up to the level of the hexkit provider, which publishes the event to the DLQ along with the error information. If the DLQ is not enabled, the error is re-raised as a `RetriesExhaustedError`. If this error is unhandled at the service level, the service will crash. In order to handle `RetriesExhaustedError` instances, the `try/except` must be placed around the `.run()` call on the subscriber. ## Protocols ### Protocols Protocols define abstract interfaces (ports) used by applications and adapters. Providers implement these protocols for specific backends. ## Supported Protocols ### KeyValueStoreProtocol The [`KeyValueStoreProtocol`](../../reference/protocols.kvstore.KeyValueStoreProtocol.qmd) defines asynchronous operations for storing and retrieving values by key. It is generic over the value type `V`, which is determined by the concrete store implementation. Keys are ordinary `str` values, i.e. normal Unicode strings. Its core operations are: - `get(key: str, default: V | None = None) -> V | None`: Retrieve a value by key. Returns `default` (or `None`) when the key is not present. - `set(key: str, value: V) -> None`: Store a value for a key. - `delete(key: str) -> None`: Remove a key/value pair if present. - `exists(key: str) -> bool`: Check whether a key exists. The protocol is implemented by multiple provider families: - in-memory testing providers: [`InMem*KeyValueStore`](../providers/testing.qmd) - Redis providers: [`Redis*KeyValueStore`](../providers/redis.qmd) - MongoDB providers: [`MongoDb*KeyValueStore`](../providers/mongodb.qmd#mongodb-kv-store-providers) - S3 providers: [`S3*KeyValueStore`](../providers/s3.qmd#s3-kv-store-providers) - Vault providers: [`Vault*KeyValueStore`](../providers/vault.qmd) Across these families, the same value-type variants are used: - `*BytesKeyValueStore`: Stores raw `bytes` values. - `*StrKeyValueStore`: Stores `str` values. - `*JsonKeyValueStore`: Stores JSON objects. - `*DtoKeyValueStore`: Stores Pydantic model instances (DTOs). This naming convention is backend-agnostic, so switching provider families does not require changing your value model at the protocol level. Choose the provider family based on the operational properties you need: in-memory for tests, Redis for fast access to small hot values, MongoDB for durable structured values, S3 for large or opaque payloads, and Vault for secrets or other sensitive values. Choose the value-type variant based on how much structure and validation you want: `Bytes` for binary or already-serialized payloads, `Str` for plain text, `Json` for flexible schema-light objects, and `Dto` for schema-validated typed models. ### Dao and DaoFactoryProtocol The [`Dao`](../../reference/protocols.dao.Dao.qmd) protocol defines asynchronous CRUD and find operations on resources described by Pydantic models, and the [`DaoFactoryProtocol`](../../reference/protocols.dao.DaoFactoryProtocol.qmd) describes the factory that constructs DAOs, one per resource type. Together they let services persist and query data without depending on a specific database. The protocol is implemented by the [MongoDB provider](../providers/mongodb.qmd#mongodb-dao) for production use and by an [in-memory mock DAO](../providers/testing.qmd) for unit testing. These interfaces are documented in depth in the dedicated [Data Access Objects](dao.qmd) chapter; for the architectural background, see [The DAO Pattern](../arch_concepts/dao_pattern.qmd). ### DaoPublisher and DaoPublisherFactoryProtocol The [`DaoPublisher`](../../reference/protocols.daopub.DaoPublisher.qmd) protocol extends the `Dao` interface with automatic event publishing: every insert, update, or delete also emits a change event, implementing the publishing side of the [outbox pattern](../arch_concepts/event_driven_arch.qmd#outbox-pattern). Instances are constructed by factories described by the [`DaoPublisherFactoryProtocol`](../../reference/protocols.daopub.DaoPublisherFactoryProtocol.qmd). The protocol is implemented by the [MongoDB + Kafka provider](../providers/mongokafka.qmd#outbox-publisher) and documented in depth in the dedicated [DAO Publisher](daopublisher.qmd) chapter. ### DaoSubscriberProtocol The [`DaoSubscriberProtocol`](../../reference/protocols.daosub.DaoSubscriberProtocol.qmd) is the consuming counterpart of the `DaoPublisher`: a translator interface for services that mirror resources owned by another service by consuming its outbox events. Implementations declare the event topic and payload model and handle just two callbacks: `changed()` and `deleted()`. Translators implementing the protocol are run by the [`KafkaOutboxSubscriber`](../providers/kafka/index.qmd) provider and documented in depth in the dedicated [DAO Subscriber](daosubscriber.qmd) chapter. ### EventPublisherProtocol and EventSubscriberProtocol The [`EventPublisherProtocol`](../../reference/protocols.eventpub.EventPublisherProtocol.qmd) and [`EventSubscriberProtocol`](../../reference/protocols.eventsub.EventSubscriberProtocol.qmd) (from `hexkit.protocols.eventpub` and `hexkit.protocols.eventsub`) define the publishing and consuming of events through an event broker: `publish()` on the one side, and translator classes with topics and types of interest plus a `consume()` method on the other. They are the foundation of hexkit's [event-driven architecture](../arch_concepts/event_driven_arch.qmd) support. Both are implemented by the [Apache Kafka provider](../providers/kafka/index.qmd); see the [Kafka Event Publisher](../providers/kafka/publisher.qmd) and [Kafka Event Subscriber](../providers/kafka/subscriber.qmd) chapters, which also document the protocol usage itself. ### ObjectStorageProtocol The [`ObjectStorageProtocol`](../../reference/protocols.objstorage.ObjectStorageProtocol.qmd) defines asynchronous operations for S3-like object storage: managing buckets and file objects, and orchestrating uploads and downloads through presigned URLs, so that file content never flows through the service itself. The protocol is implemented by the [S3 provider](../providers/s3.qmd#s3-object-storage-provider), which works with any S3-compatible backend such as AWS S3, MinIO, or Ceph RGW. This interface is documented in depth in the dedicated [Object Storage](objstorage.qmd) chapter. ## Adding Protocols TBD ### Data Access Objects The DAO protocol, defined in `hexkit.protocols.dao`, is hexkit's abstract interface for database access. It lets application code create, read, update, delete, and search resources without knowing which database sits behind it. This chapter documents the protocol itself; for the architectural background see [The DAO Pattern](../arch_concepts/dao_pattern.qmd), and for the production implementation see the [MongoDB DAO provider](../providers/mongodb.qmd#mongodb-dao). The protocol consists of two interfaces that work together: - [`Dao`](../../reference/protocols.dao.Dao.qmd) describes the operations available on a single resource type. It is a structural type (a `typing.Protocol`), so any class with matching methods satisfies it, no inheritance required. - [`DaoFactoryProtocol`](../../reference/protocols.dao.DaoFactoryProtocol.qmd) is the abstract base class that providers implement. Its `get_dao()` method constructs DAO instances, one per resource type. All operations are asynchronous and fully typed: a `Dao[Book]` accepts and returns `Book` instances, and type checkers will hold you to that. ## DTO Models Every DAO is bound to a *data transfer object* (DTO) model: a Pydantic `BaseModel` subclass describing the shape of the resources it manages. The DTO carries all fields of the resource, including its ID: ```python from pydantic import BaseModel, ConfigDict, UUID4 from hexkit.protocols.dao import UUID4Field class Book(BaseModel): """A book in our catalog.""" model_config = ConfigDict(frozen=True) id: UUID4 = UUID4Field(description="The ID of the book.") title: str author: str pages: int ``` A few things are worth noting here: - **The ID lives on the model.** There is no separate "generated key" mechanism: you put the ID on the DTO before inserting. The [`UUID4Field`](../../reference/index.qmd#protocols-data-access-objects) helper (a `pydantic.Field` preconfigured with a UUID4 default factory) makes self-generated IDs a one-liner. - **The ID field can have any name.** You tell the factory which field serves as the ID via the `id_field` argument. Its type must translate to a string or integer in the model's JSON schema, so `str`, `int`, and UUID types all qualify. - **Freezing the model is recommended.** DAOs treat DTOs as value objects that are written and read as a whole. An immutable model avoids accidental in-place edits that never reach the database; to modify a resource, create an updated copy with `model_copy(update=...)` and pass it to the DAO. ## Obtaining a DAO: the Factory DAOs are constructed by a factory that implements [`DaoFactoryProtocol`](../../reference/protocols.dao.DaoFactoryProtocol.qmd). Which factory you instantiate decides the storage backend; everything after that call is backend-independent: ```python book_dao = await dao_factory.get_dao( name="books", # the resource type, think "table" or "collection" name dto_model=Book, id_field="id", indexes=None, # optionally, provider-specific index definitions ) ``` The factory validates its inputs before handing them to the provider and raises descriptive errors when something is off: - `IdFieldNotFoundError` if the DTO model has no field named `id_field` - `IdTypeNotSupportedError` if the ID field is not string- or integer-typed - `IndexFieldsInvalidError` if an index references fields the model does not have Providers always maintain a unique index on the ID field, so IDs are guaranteed to be unique per resource type. Additional indexes can be requested through the `indexes` argument. Since index capabilities differ between databases, each provider brings its own index class derived from `IndexBase` (for example [`MongoDbIndex`](../providers/mongodb.qmd#defining-indexes)); the index definitions are the only provider-specific ingredient in an otherwise generic setup. ## CRUD Operations The [`Dao`](../../reference/protocols.dao.Dao.qmd) interface offers seven operations: | Operation | Description | Raises | | --------- | ----------- | ------ | | `insert(dto)` | Create a new resource. | `ResourceAlreadyExistsError`, `UniqueConstraintViolationError` | | `get_by_id(id_)` | Fetch a resource by its ID. | `ResourceNotFoundError` | | `update(dto)` | Replace an existing resource. | `ResourceNotFoundError`, `UniqueConstraintViolationError` | | `upsert(dto)` | Update the resource if it exists, insert it otherwise. | `UniqueConstraintViolationError` | | `delete(id_)` | Remove a resource by its ID. | `ResourceNotFoundError` | | `find_one(mapping=...)` | Fetch the single resource matching a filter. | `NoHitsFoundError`, `MultipleHitsFoundError`, `InvalidFindMappingError` | | `find_all(mapping=...)` | Iterate over all resources matching a filter. | `InvalidFindMappingError`, `ValueError` | The write operations are deliberately strict about existence. `insert` refuses to overwrite, `update` refuses to create, and both signal violations with exceptions rather than silent no-ops. That makes accidental data loss loud instead of quiet. When "create or replace" is genuinely what you want, say so explicitly by calling `upsert`. A typical session with a DAO looks like this: ```python book = Book(title="Refactoring", author="Martin Fowler", pages=448) await book_dao.insert(book) # Modify by copying (the model is frozen) and writing back: extended = book.model_copy(update={"pages": 464}) await book_dao.update(extended) same_book = await book_dao.get_by_id(book.id) assert same_book == extended await book_dao.delete(book.id) ``` ## Finding Resources Beyond lookups by ID, DAOs support searching by field values. Both find methods take a `mapping`: a dictionary whose keys are field names of the DTO model and whose values are the values a matching resource must have. Multiple entries must all match (a logical AND): ```python # Exactly one match expected: book = await book_dao.find_one(mapping={"title": "Refactoring"}) # Any number of matches: async for book in book_dao.find_all(mapping={"author": "Martin Fowler"}): print(book.title) # An empty mapping matches everything: all_books = await book_dao.find_all(mapping={}).to_list() ``` `find_one` enforces that exactly one resource matches and raises `NoHitsFoundError` or `MultipleHitsFoundError` otherwise. Use it when your logic depends on uniqueness; use `find_all` when zero or many hits are acceptable. Mapping keys are validated against the DTO model, so a typo in a field name raises `InvalidFindMappingError` instead of silently matching nothing. Pass mapping values using the same Python types as the corresponding model fields: a `UUID` for a UUID field, a `datetime` for a datetime field, and so on. Whether richer values (nested structures, or query operators expressed as dictionaries) are supported is up to the provider. The [MongoDB provider](../providers/mongodb.qmd#queries-and-filters), for example, passes dictionary values through to MongoDB's query language, enabling range and pattern queries. ### Pagination and Sorting `find_all` accepts three optional arguments for shaping large result sets: - `skip`: number of matching resources to pass over before yielding results - `limit`: maximum number of resources to yield (a limit of `0` is taken literally and yields nothing) - `sort`: a list of field names defining the sort order; prefix a name with `-` for descending order, so `["author", "-pages"]` sorts by author ascending, then by page count descending When paginating with `skip` and `limit`, always provide a `sort`. Without a defined order, the database is free to return matches in any order, and page boundaries may shift between calls. ### The FindResult Note that `find_all` is not awaited: it immediately returns a [`FindResult`](../../reference/protocols.dao.FindResult.qmd), and the actual database work happens lazily while you consume it. A `FindResult` is an async iterator with two extra conveniences: ```python result = book_dao.find_all( mapping={"author": "Martin Fowler"}, skip=20, limit=10, sort=["title"], ) books = await result.to_list() # collect everything into a list total = await result.total_count() # matches ignoring skip and limit ``` `total_count()` reports how many resources match the mapping in total, deliberately ignoring `skip` and `limit`, which is exactly the number you need to render pagination controls. It runs an extra count query on first call and caches the answer. Keep in mind that iterating (or calling `to_list()`) consumes the iterator, so collect the results once and reuse the list. ## Error Handling All DAO errors derive from a common base, [`DaoError`](../../reference/protocols.dao.DaoError.qmd), so a single `except DaoError` catches any database-related failure. The full hierarchy: - `DaoError`: base for everything below - `ResourceNotFoundError`: the requested ID does not exist (`get_by_id`, `update`, `delete`) - `ResourceAlreadyExistsError`: the ID is already taken (`insert`) - `UniqueConstraintViolationError`: a write would break a unique index on a field other than the ID - `FindError`: base for find-related errors - `InvalidFindMappingError`: a mapping key is not a field of the DTO model - `NoHitsFoundError`: `find_one` found nothing - `MultipleHitsFoundError`: `find_one` found more than one match - `DbTimeoutError`: a database operation timed out Providers translate their native driver errors into these types, so your error handling stays portable across backends: ```python from hexkit.protocols.dao import ResourceNotFoundError try: book = await book_dao.get_by_id(book_id) except ResourceNotFoundError: raise BookNotInCatalogError(book_id) from None ``` ## Transactions The `Dao` interface declares a `with_transaction()` class method that is intended to provide transaction-scoped DAOs through an async context manager, committing on success and rolling back on error. This is currently a forward-looking part of the interface: the shipped providers do not implement it yet and raise `NotImplementedError`. Each individual DAO operation is atomic on its own, but there is no way yet to group several operations into one transaction through the protocol. ## Available Providers Two implementations of the DAO protocol ship with hexkit: - The [MongoDB provider](../providers/mongodb.qmd#mongodb-dao) ([`MongoDbDaoFactory`](../../reference/providers.mongodb.MongoDbDaoFactory.qmd)) stores resources as documents in MongoDB collections. This is the production-grade implementation. - The [in-memory mock DAO](../providers/testing.qmd) ([`new_mock_dao_class`](../../reference/providers.testing.new_mock_dao_class.qmd)) stores resources in a plain dictionary and even understands a subset of MongoDB's query operators. It is ideal for unit tests, where it lets you exercise real DAO semantics without any database. For extensive, runnable usage examples, the [DAO integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/dao/test_mongodb_dao.py) in the hexkit repository are a good complement to this guide. ## Outbox Variants Two sibling protocols extend the DAO idea with event publishing, implementing a variation of the *outbox pattern*: - [`DaoPublisher`](daopublisher.qmd) and its factory (from `hexkit.protocols.daopub`) describe DAOs that automatically publish an event whenever a resource is inserted, updated, or deleted. - The [`DaoSubscriberProtocol`](daosubscriber.qmd) (from `hexkit.protocols.daosub`) is the consuming counterpart for services that want to mirror those changes. Both have their own chapters, linked above; the publishing side is implemented by the [MongoDB + Kafka provider](../providers/mongokafka.qmd) and further discussed in the [Publisher Variants](../developer_help/outbox_v_persistent_pub.qmd) chapter. If you only need plain persistence, you can safely ignore them; they build on everything described above. ## Implementing a Custom Provider Supporting a new database means implementing the two interfaces for it. Three rules keep a custom provider well-behaved: 1. **Subclass `DaoFactoryProtocol` and implement `_get_dao()`.** The public `get_dao()` method performs all input validation and then delegates to your `_get_dao()`, so your implementation can trust its arguments. 2. **Return objects that structurally satisfy `Dao`.** Since `Dao` is a `typing.Protocol`, your DAO class does not inherit from anything; it just needs the right methods with the right signatures. 3. **Raise the shared error types.** Translate your driver's errors into `ResourceNotFoundError`, `DbTimeoutError`, and friends, so that translators written against the protocol keep working with your provider. The following compact but functional example stores resources in plain dictionaries. A real provider would talk to a database instead, but the structure stays the same (compare with the [MongoDB implementation](https://github.com/ghga-de/hexkit/blob/main/src/hexkit/providers/mongodb/provider/dao.py)): ```python from collections.abc import AsyncIterator, Collection, Mapping from contextlib import AbstractAsyncContextManager from typing import Any, Generic from hexkit.custom_types import ID from hexkit.protocols.dao import ( Dao, DaoFactoryProtocol, Dto, FindResult, IndexBase, MultipleHitsFoundError, NoHitsFoundError, ResourceAlreadyExistsError, ResourceNotFoundError, ) class DictIndex(IndexBase): """Index description for the dict provider (declared but not acted upon).""" def __init__(self, *fields: str): self._fields = list(fields) def list_field_names(self) -> list[str]: """Return the names of all fields covered by this index.""" return self._fields class DictDao(Generic[Dto]): """A DAO keeping all resources in an in-memory dict.""" def __init__(self, *, dto_model: type[Dto], id_field: str): self._dto_model = dto_model self._id_field = id_field self._resources: dict[ID, Dto] = {} def _id_of(self, dto: Dto) -> ID: return getattr(dto, self._id_field) async def get_by_id(self, id_: ID) -> Dto: try: return self._resources[id_] except KeyError as error: raise ResourceNotFoundError(id_=id_) from error async def insert(self, dto: Dto) -> None: id_ = self._id_of(dto) if id_ in self._resources: raise ResourceAlreadyExistsError(id_=id_) self._resources[id_] = dto async def update(self, dto: Dto) -> None: id_ = self._id_of(dto) if id_ not in self._resources: raise ResourceNotFoundError(id_=id_) self._resources[id_] = dto async def upsert(self, dto: Dto) -> None: self._resources[self._id_of(dto)] = dto async def delete(self, id_: ID) -> None: if id_ not in self._resources: raise ResourceNotFoundError(id_=id_) del self._resources[id_] async def find_one(self, *, mapping: Mapping[str, Any]) -> Dto: hits = await self.find_all(mapping=mapping).to_list() if not hits: raise NoHitsFoundError(mapping=mapping) if len(hits) > 1: raise MultipleHitsFoundError(mapping=mapping) return hits[0] def find_all( self, *, mapping: Mapping[str, Any], skip: int | None = None, limit: int | None = None, sort: list[str] | None = None, # sorting omitted for brevity ) -> FindResult[Dto]: hits = [ dto for dto in self._resources.values() if all(getattr(dto, field) == value for field, value in mapping.items()) ] end = None if limit is None else (skip or 0) + limit page = hits[skip or 0 : end] async def _iterate() -> AsyncIterator[Dto]: for dto in page: yield dto async def _total_count() -> int: return len(hits) return FindResult(results_iterator=_iterate(), get_total_count=_total_count) @classmethod def with_transaction(cls) -> AbstractAsyncContextManager["Dao[Dto]"]: raise NotImplementedError() class DictDaoFactory(DaoFactoryProtocol[DictIndex]): """A DAO factory producing dict-backed DAOs.""" def __init__(self) -> None: self._daos: dict[str, DictDao[Any]] = {} async def _get_dao( self, *, name: str, dto_model: type[Dto], id_field: str, indexes: Collection[DictIndex] | None, ) -> Dao[Dto]: """Called by get_dao() after input validation, so arguments can be trusted.""" if name not in self._daos: self._daos[name] = DictDao(dto_model=dto_model, id_field=id_field) return self._daos[name] ``` With this in place, the factory is a drop-in replacement anywhere a `DaoFactoryProtocol` is expected: ```python dao_factory = DictDaoFactory() book_dao = await dao_factory.get_dao(name="books", dto_model=Book, id_field="id") await book_dao.insert(Book(title="Refactoring", author="Martin Fowler", pages=448)) ``` If you build a provider for a widely used database, please consider [contributing it to hexkit](https://github.com/ghga-de/hexkit). ### DAO Publisher The DAO publisher protocol, defined in `hexkit.protocols.daopub`, is the publishing half of hexkit's [outbox pattern](../arch_concepts/event_driven_arch.qmd#outbox-pattern): a [DAO](dao.qmd) that automatically publishes an event for every insert, update, and delete. Services that own data use it in place of a plain DAO; other services mirror the data by consuming the resulting event stream through the [DAO Subscriber](daosubscriber.qmd), the consuming counterpart. The appeal of the pattern is that the database remains the single source of truth. Events are derived from persisted state, and can be derived from it *again*: if a publish fails, if a topic is lost, or if a new consumer needs the full data set, the events are simply re-emitted from the database. Like the plain DAO protocol, this one consists of two interfaces: - [`DaoPublisher`](../../reference/protocols.daopub.DaoPublisher.qmd) describes the DAO handed to application code. It is a structural type (a `typing.Protocol`) extending [`Dao`](dao.qmd) with two publishing methods. - [`DaoPublisherFactoryProtocol`](../../reference/protocols.daopub.DaoPublisherFactoryProtocol.qmd) is the abstract base class that providers implement; its `get_dao()` constructs DAO publishers, one per resource type. ## The `DaoPublisher` Interface A `DaoPublisher[Dto]` offers everything a `Dao[Dto]` does — the full set of [CRUD and find operations](dao.qmd#crud-operations) with identical semantics and [error types](dao.qmd#error-handling) — so it drops into any code written against the DAO protocol. Publishing happens as a side effect of the write operations; the calling code does not deal with events at all. On top of the DAO surface, two methods control publishing explicitly: - `publish_pending()`: publish the changes that have not been published yet — for example because a publish failed, or because automatic publishing is turned off. - `republish()`: re-emit the current state of *all* resources, regardless of whether it was published before. Both are idempotent from the consumers' perspective (consuming the same state twice must be safe anyway — see [below](#event-semantics)). Services typically expose them through a CLI command that operators run as a job, for instance to backfill a newly created consumer service; the [Publisher Variants](../developer_help/outbox_v_persistent_pub.qmd#republishing) chapter shows how to structure such a command. ## The Factory DAO publishers are constructed by a factory implementing [`DaoPublisherFactoryProtocol`](../../reference/protocols.daopub.DaoPublisherFactoryProtocol.qmd). Its `get_dao()` method takes the same arguments as the [plain DAO factory](dao.qmd#obtaining-a-dao-the-factory) — `name`, `dto_model`, `id_field`, and optional `indexes` — plus three that define the publishing behavior: ```python user_dao = await dao_publisher_factory.get_dao( name="users", dto_model=User, id_field="user_id", dto_to_event=lambda user: user.model_dump(mode="json", exclude={"internal_notes"}), event_topic="users", autopublish=True, ) ``` - `dto_to_event`: a function turning a DTO into the event payload. It decouples what you *store* from what you *expose*: strip internal fields, rename, or reshape as needed. Returning `None` suppresses the event for that change entirely, which allows publishing only a subset of resources. - `event_topic`: the topic the events are published to. Use one dedicated topic per resource type, as consumers subscribe to the topic as a whole. - `autopublish`: whether every write publishes immediately (the default). When `False`, changes are only persisted, and events are emitted later by an explicit `publish_pending()` call. The factory validates its inputs like the plain DAO factory does and raises the same errors (`IdFieldNotFoundError`, `IdTypeNotSupportedError`). ## Event Semantics The events form a *changelog* of the resource collection, with a deliberately minimal contract that is hardcoded in hexkit rather than configurable: - Every event's *key* is the resource ID (the value of the DTO's `id_field`), so all changes to one resource share a key and stay ordered within a partition. - An insert, update, or upsert produces an event of type **`upserted`** whose payload is the full state of the resource as returned by `dto_to_event`. Events do not describe *what changed* — they carry the entire latest state. - A delete produces an event of type **`deleted`** with an empty payload — a *tombstone* whose key tells consumers which resource is gone. Because every event carries the full state, only the latest event per key matters. This makes the topics ideal candidates for [Kafka topic compaction](../arch_concepts/event_driven_arch.qmd#topic-compaction-idempotence): the broker retains only the newest event per resource, and a new consumer reading the compacted topic from the beginning receives exactly the current data set. Consuming these events is the subject of the [DAO Subscriber](daosubscriber.qmd) chapter. ## Failure Model The point of the outbox pattern is that database write and event publish do not need to succeed atomically. The state change is persisted along with a flag tracking whether the latest change has been published; when publishing fails, the data is safe and the event is merely *pending*. A later `publish_pending()` — whether run manually, scheduled as a job, or triggered on service startup — brings the event stream back in line with the database. Two identity details are worth knowing for tracing and debugging: - The *correlation ID* active during the original write is stored with the resource and re-applied when its events are published or republished, so traces stay consistent even for events re-emitted much later. - *Event IDs* are freshly generated on every publish and republish: re-emissions are distinct events (with the same key and payload), not byte-identical copies. This differs from the [`PersistentKafkaPublisher`](../providers/mongokafka.qmd#persistent-publisher), which reuses stored event IDs. ## Available Providers One implementation ships with hexkit: [`MongoKafkaDaoPublisherFactory`](../../reference/providers.mongokafka.MongoKafkaDaoPublisherFactory.qmd) from the [MongoDB + Kafka provider](../providers/mongokafka.qmd#outbox-publisher) (requiring the `mongodb` and `akafka` extras). It stores resources in MongoDB with a small `__metadata__` field (publish status, correlation ID, deletion flag) and publishes through Kafka; deleted resources leave a metadata stub behind so that deletions survive for republishing. The provider chapter documents these mechanics, along with a complete usage example. For runnable examples, see the [mongokafka integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/daopubsub/test_mongokafka_pubsub.py) in the hexkit repository. ## Implementing a Custom Provider Supporting another database/broker combination means subclassing `DaoPublisherFactoryProtocol` and implementing `_get_dao()`; the public `get_dao()` validates all inputs first, so the implementation can trust its arguments. The rules from [implementing a custom DAO provider](dao.qmd#implementing-a-custom-provider) apply unchanged, plus two specific to publishing: 1. **Honor the event contract.** Publish `upserted` events with the `dto_to_event` payload and `deleted` tombstones, keyed by resource ID, so [`DaoSubscriberProtocol`](daosubscriber.qmd) consumers remain compatible with your provider. 2. **Persist the publish state.** `publish_pending()` and `republish()` must be able to reconstruct and re-emit events from stored state at any time — that is what makes it an outbox rather than plain publishing. If you build a provider for a widely used backend combination, please consider [contributing it to hexkit](https://github.com/ghga-de/hexkit). ### DAO Subscriber The [`DaoSubscriberProtocol`](../../reference/protocols.daosub.DaoSubscriberProtocol.qmd), defined in `hexkit.protocols.daosub`, is the consuming half of hexkit's [outbox pattern](../arch_concepts/event_driven_arch.qmd#outbox-pattern): a translator interface for services that keep their own copy of resources owned by another service, kept up to date by consuming the change events that the owning service publishes through a [DAO Publisher](daopublisher.qmd). Where a generic event translator (the [`EventSubscriberProtocol`](../providers/kafka/subscriber.qmd#the-translator)) receives raw events and decides everything itself, a DAO subscriber works at a higher level: the provider already knows the outbox event contract, validates the payload against your model, and calls one of exactly two methods — *this resource changed* or *this resource was deleted*. ## Use Case: Syncing Data Across Services In a microservice landscape, service B often needs data that service A owns — say, a notification service needing user contact details from the user registry. Direct API calls would couple the services and make B's availability depend on A's. The outbox pattern instead replicates the data: A publishes every state change of its resources to a Kafka topic, and B consumes that topic to maintain its own local copy, which it can then query freely. The [Event-Driven Architecture](../arch_concepts/event_driven_arch.qmd#outbox-pattern) chapter discusses the trade-offs. ```{mermaid} %%| fig-cap: Service B mirrors resources owned by service A by consuming its outbox events. sequenceDiagram autonumber participant CoreA as Service A core participant Outbox as DAO publisher (A) participant Kafka as Kafka topic "users" participant Sub as DAO subscriber (B) participant DaoB as Local DAO (B) CoreA->>Outbox: upsert(user) Outbox->>Kafka: event type "upserted", key=user_id, full state Kafka->>Sub: changed(resource_id, dto) Sub->>DaoB: upsert local copy CoreA->>Outbox: delete(user_id) Outbox->>Kafka: event type "deleted", key=user_id (tombstone) Kafka->>Sub: deleted(resource_id) Sub->>DaoB: delete local copy ``` The event stream a DAO subscriber consumes follows the outbox contract described in the [DAO Publisher](daopublisher.qmd#event-semantics) chapter: one topic per resource type, the resource ID as the event key, and just two event types — `upserted`, carrying the full latest state of the resource, and `deleted`, an empty-bodied tombstone. There is no "what changed" delta to interpret; consumers simply overwrite or drop their local copy. ## Implementing a Subscriber `DaoSubscriberProtocol` is an abstract base class, generic over the DTO model. An implementation provides two class attributes and two methods: - `event_topic`: the outbox topic to consume from. - `dto_model`: a Pydantic model describing the expected event payload. The *provider* validates each incoming payload against it, so your code always receives a typed, validated object. - `changed(resource_id, update)`: called for `upserted` events with the validated DTO. One method covers both creation and update — implement it as an upsert. - `deleted(resource_id)`: called for `deleted` events. A complete example that mirrors user data into a local [DAO](dao.qmd): ```python from pydantic import BaseModel from hexkit.protocols.dao import Dao, ResourceNotFoundError from hexkit.protocols.daosub import DaoSubscriberProtocol class User(BaseModel): """The subset of user data published by the user registry.""" user_id: str name: str email: str class UserOutboxSubscriber(DaoSubscriberProtocol[User]): """Keeps a local copy of the user registry's data in sync.""" event_topic = "users" dto_model = User def __init__(self, *, user_dao: Dao[User]): self._user_dao = user_dao async def changed(self, resource_id: str, update: User) -> None: """Upsert the local copy of the user.""" await self._user_dao.upsert(update) async def deleted(self, resource_id: str) -> None: """Remove the local copy of the user, tolerating that it may not exist.""" try: await self._user_dao.delete(resource_id) except ResourceNotFoundError: pass # already absent, nothing to do ``` Note that the `dto_model` describes the *event payload* as published by the owning service (its `dto_to_event` output), which is not necessarily the model the owning service stores. The two services typically share the payload schema through a dedicated contract library. ### Implementations Must Be Idempotent Outbox events are, by design, re-emittable: the owning service can [republish](daopublisher.qmd#the-daopublisher-interface) its data at any time, and Kafka's at-least-once delivery can redeliver events after failures. Your implementation must therefore tolerate consuming the same event multiple times — which the example above achieves naturally by upserting on `changed` and treating a missing resource in `deleted` as success. Republishing also means events may arrive *out of order* relative to their original emission (though changes to a single resource keep their order within a topic partition, since they share the event key). Avoid logic that depends on cross-resource sequencing, and lean on [topic compaction](../arch_concepts/event_driven_arch.qmd#topic-compaction-idempotence) where the latest state per resource is all that matters. ## Running a Subscriber: the Kafka Provider The [`KafkaOutboxSubscriber`](../../reference/providers.akafka.KafkaOutboxSubscriber.qmd) runs one or more DAO subscribers against Kafka. It accepts a sequence of translators — at most one per topic — and is configured and operated like the [`KafkaEventSubscriber`](../providers/kafka/subscriber.qmd) it wraps: ```python from hexkit.providers.akafka import ( KafkaConfig, KafkaEventPublisher, KafkaOutboxSubscriber, ) config = KafkaConfig( service_name="notification-service", service_instance_id="instance-001", kafka_servers=["localhost:9092"], kafka_enable_dlq=True, ) user_subscriber = UserOutboxSubscriber(user_dao=user_dao) async with ( KafkaEventPublisher.construct(config=config) as dlq_publisher, KafkaOutboxSubscriber.construct( config=config, translators=[user_subscriber], dlq_publisher=dlq_publisher, ) as outbox_subscriber, ): await outbox_subscriber.run() ``` `run()` consumes indefinitely; `run(forever=False)` handles a single event, which is convenient in tests. As with the plain subscriber, a `dlq_publisher` is required whenever the dead letter queue is enabled. Under the hood, the provider wraps the translators in a [`ComboTranslator`](../../reference/providers.akafka.ComboTranslator.qmd) and drives a regular `KafkaEventSubscriber` with it. This has two practical consequences: - All of the plain subscriber's machinery applies as described in [its chapter](../providers/kafka/subscriber.qmd): consumer groups, offset management with at-least-once delivery, correlation ID propagation, retries, and the DLQ. - A `ComboTranslator` can mix DAO subscribers with plain `EventSubscriberProtocol` translators, so a service consuming both outbox and regular events can run everything through one `KafkaEventSubscriber` instead of a separate `KafkaOutboxSubscriber`. ## Error Handling The division of labor: payload validation is the provider's job, everything after that is yours. - If an `upserted` payload does not validate against `dto_model`, the provider raises a [`DtoValidationError`](../../reference/protocols.daosub.DtoValidationError.qmd) (defined alongside the protocol in `hexkit.protocols.daosub`); `changed()` is never called with malformed data. Custom providers consuming outbox events are expected to follow the same convention. - Exceptions raised inside `changed()` or `deleted()` — as well as the `DtoValidationError` itself — flow into the standard [failure handling](../providers/kafka/subscriber.qmd#failure-handling-retries-and-the-dlq) of the underlying subscriber: configurable immediate retries, then either the dead letter queue or a service crash without committing the offset. Persistent validation failures usually indicate a contract mismatch between the services and end up in the DLQ for inspection, without stalling the consumption of other events. ## Testing The [Kafka test utils](../providers/kafka/test_utils.qmd) cover the outbox case as well: publish an `upserted` or `deleted` event with `publish_event()` (using the resource ID as `key`) and let a `KafkaOutboxSubscriber` consume it with `run(forever=False)`. For runnable examples, see the [outbox subscriber integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/daopubsub/test_akafka_daosub.py) in the hexkit repository. ### Object Storage The [`ObjectStorageProtocol`](../../reference/protocols.objstorage.ObjectStorageProtocol.qmd), defined in `hexkit.protocols.objstorage`, is hexkit's abstract interface for S3-like object storage: *buckets* that hold *file objects*, both addressed by string IDs. It lets application code manage buckets, orchestrate uploads and downloads, and inspect or delete objects without knowing which storage backend sits behind it. This chapter documents the protocol itself; for the production implementation see the [S3 provider](../providers/s3.qmd#s3-object-storage-provider). Unlike the [DAO protocol](dao.qmd), which is partly structural, this is a classic abstract base class: providers subclass `ObjectStorageProtocol` and implement its abstract methods. All operations are asynchronous. ## Design: The Service Never Touches File Content The most important thing to understand about this protocol is what it does *not* offer: there is no `upload(content)` and no `download() -> bytes`. File content never flows through your application. Instead, the service *orchestrates* transfers using *presigned URLs*: it asks the storage for a temporary, pre-authorized URL, hands that URL to the client (or another service), and the actual bytes travel directly between the client and the storage backend. Afterwards, the service can verify the outcome through metadata operations like `does_object_exist`, `get_object_size`, or `get_object_etag`. A file upload, for example, plays out like this: ```{mermaid} %%| fig-cap: An upload orchestrated via a presigned URL; the file content (5) travels directly between client and storage. sequenceDiagram autonumber participant Client participant Service as Your service participant Storage as Object storage Client->>Service: request file upload Service->>Storage: get_object_upload_url(...) Storage-->>Service: presigned URL Service-->>Client: presigned URL Client->>Storage: POST file content (direct transfer) Client->>Service: report completion Service->>Storage: does_object_exist(...), get_object_size(...) ``` This design has far-reaching consequences: - **Services stay stateless and memory-light.** Whether a file is 5 KiB or 5 TiB, the service only ever handles small metadata and URLs. - **Transfer traffic bypasses the application.** Bandwidth-heavy uploads and downloads scale with the storage backend, not with your service. - **URLs are credentials.** A presigned URL grants time-limited access to exactly one operation on exactly one object, without sharing storage credentials. Every URL has an expiry (`expires_after`, in seconds); the default for uploads and downloads is 24 hours (the `DEFAULT_URL_EXPIRATION_PERIOD` constant). Uploads come in two flavors: a *presigned POST* for uploading an object in one request, and a *multipart upload* that splits large files into parts, each uploaded through its own presigned URL. Both are covered below. ## Bucket and Object IDs Every public method validates the given bucket and object IDs before contacting the backend, so malformed IDs fail fast and with a clear message, no matter the provider. The rules are modeled on AWS naming restrictions: | ID kind | Rules | Raised on violation | | ------- | ----- | ------------------- | | Bucket ID | 3-63 lowercase letters, digits, or hyphens; may not start or end with a hyphen | `BucketIdValidationError` | | Object ID | 3-255 letters, digits, hyphens, underscores, or dots; may not start or end with a hyphen or a dot | `ObjectIdValidationError` | Note that the default rules do not allow forward slashes in object IDs, which are sometimes used to emulate directories. Providers may relax the rules where the backend supports more: the [S3 provider](../providers/s3.qmd#ceph-tenant-buckets), for example, additionally accepts Ceph's tenant-prefixed bucket names. ## Bucket Operations Buckets are flat containers for objects and are managed with four operations: ```python if not await storage.does_bucket_exist("inbox"): await storage.create_bucket("inbox") # BucketAlreadyExistsError if it exists object_ids = await storage.list_all_object_ids("inbox") # Refuses to delete a non-empty bucket (raises BucketNotEmptyError): await storage.delete_bucket("inbox") # ...unless the content deletion is requested explicitly: await storage.delete_bucket("inbox", delete_content=True) ``` As with the write operations of the [DAO protocol](dao.qmd#crud-operations), destructive operations are deliberately loud: deleting a bucket that still holds objects fails with a `BucketNotEmptyError` unless you explicitly pass `delete_content=True`. ## Uploading Objects ### Single-Object Uploads via Presigned POST For files that fit into one request, `get_object_upload_url` returns a [`PresignedPostURL`](../../reference/protocols.objstorage.PresignedPostURL.qmd), a named tuple carrying the target `url` and a dictionary of `fields` that must accompany the POST request as form data: ```python # Fails with ObjectAlreadyExistsError if the object exists, # or with BucketNotFoundError if the bucket does not: presigned_post = await storage.get_object_upload_url( bucket_id="inbox", object_id="sequencing-data-001", expires_after=3600, # URL validity in seconds (default: 24 hours) max_upload_size=50 * 1024**2, # optionally cap the upload at 50 MiB ) ``` The service then passes `presigned_post.url` and `presigned_post.fields` to whoever performs the upload. The uploading side sends a multipart-form POST request, for example with `httpx`: ```python import httpx with open("sequencing-data.fastq", "rb") as file: response = httpx.post( presigned_post.url, data=presigned_post.fields, files={"file": file}, ) response.raise_for_status() ``` The optional `max_upload_size` (in bytes) is enforced by the storage backend itself: requests exceeding it are rejected at upload time, protecting you from oversized uploads even though the transfer never touches your service. ### Multipart Uploads Large files are uploaded in parts. The service initiates the upload, hands out one presigned URL per part, and finally completes (or aborts) the whole procedure: ```python # Initiate (fails with MultiPartUploadAlreadyExistsError if another # upload is already active for this object): upload_id = await storage.init_multipart_upload( bucket_id="inbox", object_id="sequencing-data-002" ) part_size = 16 * 1024**2 # e.g. the DEFAULT_PART_SIZE of 16 MiB # Hand out one URL per part; the uploading side sends each part # as an HTTP PUT request to its URL: for part_number in range(1, number_of_parts + 1): part_url = await storage.get_part_upload_url( upload_id=upload_id, bucket_id="inbox", object_id="sequencing-data-002", part_number=part_number, expires_after=3600, # part URLs default to 1 hour ) ... # transfer the URL to the uploading party # Complete the upload, verifying the parts that actually arrived # (fails with MultiPartUploadConfirmError if they don't add up): await storage.complete_multipart_upload( upload_id=upload_id, bucket_id="inbox", object_id="sequencing-data-002", anticipated_part_quantity=number_of_parts, anticipated_part_size=part_size, ) ``` The rules of the game: - **Part numbers** start at 1 and may not exceed 10,000 (the `MAX_FILE_PART_NUMBER` constant); parts should be uploaded in sequence. - **All parts except the last** must have the same size, and the last part may not be larger than the others. Passing `anticipated_part_quantity` and `anticipated_part_size` to `complete_multipart_upload` makes the completion verify that exactly the expected parts arrived. - **One active upload per object.** Initiating a second multipart upload for the same object raises a `MultiPartUploadAlreadyExistsError`. - **Aborting cleans up.** `abort_multipart_upload` cancels the procedure and deletes all content uploaded so far; it raises a `MultiPartUploadAbortError` if parts could not be removed (for example because a part upload was still running). - **Integrity checks are optional but available.** `get_part_upload_url` accepts a `part_md5` checksum that the backend verifies when the part arrives. Choosing a part size involves trade-offs between part count limits and request overhead; the `calc_part_size` helper [documented with the S3 provider](../providers/s3.qmd#multipart-uploads-and-part-sizing) recommends a suitable value for a given file size. ### Inspecting Multipart Uploads Three operations let you observe ongoing multipart uploads, for example to resume or clean up after interruptions: - `list_multipart_uploads_for_object(bucket_id=..., object_id=...)` returns the IDs of all active uploads for one object. - `get_all_multipart_uploads(bucket_id=...)` returns a dictionary mapping upload IDs to object IDs for a whole bucket. - `list_parts(bucket_id=..., object_id=..., upload_id=...)` lists the parts uploaded so far as dictionaries with `PartNumber`, `Size`, `ETag`, and `LastModified` keys; `max_parts` and `first_part_no` narrow the selection. Since the protocol expects one active upload per object, operations that resolve an upload by object may raise a `MultipleActiveUploadsError` when they encounter several, a state you can resolve by aborting the stale uploads. ## Downloading Objects Downloads mirror the upload design: `get_object_download_url` returns a presigned URL (a plain string this time) from which the object can be fetched with an ordinary HTTP GET request: ```python # Fails with ObjectNotFoundError if the object does not exist: download_url = await storage.get_object_download_url( bucket_id="outbox", object_id="sequencing-data-001", expires_after=3600, # URL validity in seconds (default: 24 hours) ) ``` ## Object Management The remaining operations inspect, copy, and delete objects without transferring their content: | Operation | Description | Raises | | --------- | ----------- | ------ | | `does_object_exist(bucket_id=..., object_id=...)` | Check whether an object exists. | — | | `get_object_etag(bucket_id=..., object_id=...)` | Return the object's etag, useful for change detection. | `ObjectNotFoundError` | | `get_object_size(bucket_id=..., object_id=...)` | Return the object's size in bytes. | `ObjectNotFoundError` | | `copy_object(source_bucket_id=..., source_object_id=..., dest_bucket_id=..., dest_object_id=...)` | Copy an object to another bucket and/or ID, server-side. | `ObjectNotFoundError`, `ObjectAlreadyExistsError` | | `delete_object(bucket_id=..., object_id=...)` | Delete an object. | `ObjectNotFoundError` | A typical post-upload sequence: ```python assert await storage.does_object_exist( bucket_id="inbox", object_id="sequencing-data-001" ) size = await storage.get_object_size( bucket_id="inbox", object_id="sequencing-data-001" ) # Move the object to its final destination (copy + delete): await storage.copy_object( source_bucket_id="inbox", source_object_id="sequencing-data-001", dest_bucket_id="archive", dest_object_id="sequencing-data-001", ) await storage.delete_object(bucket_id="inbox", object_id="sequencing-data-001") ``` `copy_object` copies within the storage backend; the content does not pass through your service. Backends may internally use a multipart procedure for large copies, so a failed copy can leave an unfinished multipart upload behind; with the default `abort_failed=True`, the operation cleans that up automatically. Only set `abort_failed=False` when other multipart operations may be running concurrently for the same destination object, so an unrelated upload is not aborted by accident. `does_object_exist` also accepts an `object_md5sum` argument intended for verifying content integrity, but note that the S3 provider does not implement this check yet and raises a `NotImplementedError` when it is passed. ## Error Handling All errors derive from a common base, `ObjectStorageProtocolError` (a subclass of `RuntimeError`), splitting into a bucket branch and an object branch: - `ObjectStorageProtocolError`: base for everything below - `BucketError`: base for bucket-related errors - `BucketNotFoundError`: the specified bucket does not exist - `BucketAlreadyExistsError`: `create_bucket` on an existing bucket - `BucketNotEmptyError`: `delete_bucket` on a non-empty bucket without `delete_content=True` - `BucketIdValidationError`: the bucket ID violates the naming rules - `ObjectError`: base for object-related errors - `ObjectNotFoundError`: the specified object does not exist - `ObjectAlreadyExistsError`: the target object ID is already taken - `ObjectIdValidationError`: the object ID violates the naming rules - `MultiPartUploadError`: base for multipart upload errors - `MultiPartUploadAlreadyExistsError`: another upload is already active for the object - `MultipleActiveUploadsError`: several active uploads were found where one was expected - `MultiPartUploadNotFoundError`: no upload with the given ID exists for the object - `MultiPartUploadConfirmError`: `complete_multipart_upload` rejected the upload (missing parts, unexpected sizes) - `MultiPartUploadAbortError`: `abort_multipart_upload` could not remove all parts In contrast to the DAO protocol, these exception classes are *nested* on the `ObjectStorageProtocol` class rather than defined at module level, so you access them via the class: ```python from hexkit.protocols.objstorage import ObjectStorageProtocol try: await storage.delete_object(bucket_id="inbox", object_id=object_id) except ObjectStorageProtocol.ObjectNotFoundError: ... # already gone, nothing to do ``` Providers translate their native errors into these types, so error handling written against the protocol stays portable across backends. ## Available Providers One implementation of the protocol ships with hexkit: - The [S3 provider](../providers/s3.qmd#s3-object-storage-provider) ([`S3ObjectStorage`](../../reference/providers.s3.S3ObjectStorage.qmd)) talks to any storage exposing the S3 API: AWS S3, MinIO, Ceph RGW, LocalStack, and others. For testing, the [S3 test utils](../providers/s3.qmd#s3-object-storage-test-utils) provide pytest fixtures that spin up a disposable, LocalStack-backed object storage, so integration tests exercise the real protocol semantics without external infrastructure. For extensive, runnable usage examples, the [object storage integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/objstorage/test_s3_objstorage.py) in the hexkit repository are a good complement to this guide. ## Implementing a Custom Provider Supporting a new storage backend means subclassing `ObjectStorageProtocol`. Three rules keep a custom provider well-behaved: 1. **Implement the underscore-prefixed abstract methods.** For every public operation there is an abstract counterpart (`_create_bucket`, `_get_object_upload_url`, and so on). The public methods perform ID validation and then delegate, so your implementations can trust their inputs. 2. **Raise the nested protocol exceptions.** Translate your backend's errors into `ObjectStorageProtocol.BucketNotFoundError` and friends, so error handling written against the protocol keeps working with your provider. 3. **Keep the class constants unchanged.** `DEFAULT_URL_EXPIRATION_PERIOD`, `DEFAULT_PART_SIZE`, and `MAX_FILE_PART_NUMBER` are part of the contract. The ID validation rules, on the other hand, may be relaxed by overriding the `_re_bucket_id` / `_re_object_id` patterns or the validation classmethods, as the S3 provider does for tenant-prefixed bucket names. The [S3 implementation](https://github.com/ghga-de/hexkit/blob/main/src/hexkit/providers/s3/provider/objstorage.py) serves as the reference. To check a new provider for protocol compliance, the `typical_workflow` helper from the [S3 test utils](../providers/s3.qmd#s3-object-storage-test-utils) runs a complete create-upload-copy-download-delete scenario against any `ObjectStorageProtocol` implementation. If you build a provider for a widely used storage backend, please consider [contributing it to hexkit](https://github.com/ghga-de/hexkit). ## Providers ### Providers Providers are hexkit's concrete implementations of its [protocols](../protocols/index.qmd), each binding a protocol to a specific technology. Application code is written against the protocols; which provider is wired in decides the actual infrastructure. Each provider family lives behind an optional extra (e.g. `pip install hexkit[akafka,mongodb]`), so services only pull in the dependencies they need. ## Currently Implemented Providers - [Apache Kafka](kafka/index.qmd): event publishing and consumption — [`KafkaEventPublisher`](kafka/publisher.qmd), [`KafkaEventSubscriber`](kafka/subscriber.qmd), and the `KafkaOutboxSubscriber` — plus [test utilities](kafka/test_utils.qmd). - [MongoDB](mongodb.qmd): document-database implementations of the DAO and key-value-store protocols, plus [migration tooling](mongodb_migrations.qmd) for evolving stored data alongside the data model. - [MongoDB + Kafka](mongokafka.qmd): the outbox publisher (`MongoKafkaDaoPublisher`) and the `PersistentKafkaPublisher`, combining both backends. - [S3](s3.qmd): object storage and key-value stores on any S3-compatible backend. - [Redis](redis.qmd): key-value stores for fast access to hot values. - [Vault](vault.qmd): key-value stores for secrets. - [Testing](testing.qmd): in-memory implementations for unit tests, such as the mock DAO and in-memory key-value stores. ## Adding Providers ### Apache Kafka The `akafka` subpackage (`hexkit.providers.akafka`) implements hexkit's event protocols on top of [Apache Kafka](https://kafka.apache.org/), using the asynchronous [aiokafka](https://aiokafka.readthedocs.io/) client library. It is hexkit's production event-streaming backend; the architectural background is covered in the [Event-Driven Architecture](../../arch_concepts/event_driven_arch.qmd) chapter. The subpackage provides: - [`KafkaEventPublisher`](publisher.qmd) — publishes events, implementing the [`EventPublisherProtocol`](../../../reference/protocols.eventpub.EventPublisherProtocol.qmd). - [`KafkaEventSubscriber`](subscriber.qmd) — consumes events and hands them to a service-defined translator implementing the [`EventSubscriberProtocol`](../../../reference/protocols.eventsub.EventSubscriberProtocol.qmd). - [`KafkaOutboxSubscriber`](../../protocols/daosubscriber.qmd#running-a-subscriber-the-kafka-provider) — a convenience wrapper for consuming outbox events with translators implementing the [`DaoSubscriberProtocol`](../../protocols/daosubscriber.qmd). - [Kafka test utils](test_utils.qmd) — pytest fixtures and helpers for testing against a real, disposable Kafka broker. Two hybrid providers combine Kafka with MongoDB and are documented in the [MongoDB + Kafka](../mongokafka.qmd) chapter: the outbox publisher (`MongoKafkaDaoPublisher`) and the `PersistentKafkaPublisher`. Install hexkit with the `akafka` extra to use these providers: ```bash pip install hexkit[akafka] ``` ## Configuration All Kafka providers are configured through a single Pydantic settings class, [`KafkaConfig`](../../../reference/providers.akafka.KafkaConfig.qmd) (`hexkit.providers.akafka.KafkaConfig`). The connection-related parameters, relevant to every Kafka provider, are: | Parameter | Default | Description | | --------- | ------- | ----------- | | `service_name` | *required* | Name of the service, e.g. `"user-registry"`. Doubles as the Kafka *consumer group ID*, so all instances of a service share the consumption workload. | | `service_instance_id` | *required* | Identifier unique to this instance of the service, e.g. `"instance-001"`. The Kafka *client ID* is derived as `.`. | | `kafka_servers` | *required* | List of connection strings for the Kafka bootstrap servers, e.g. `["localhost:9092"]`. | | `kafka_security_protocol` | `"PLAINTEXT"` | Either `"PLAINTEXT"` or `"SSL"`. | | `kafka_ssl_cafile` | `""` | Path to the certificate authority file used to sign the broker certificates. If empty, the system CA is used if OpenSSL finds one. | | `kafka_ssl_certfile` | `""` | Optional client certificate file (including any CA certificates needed to establish its authenticity). | | `kafka_ssl_keyfile` | `""` | Optional client private key file. | | `kafka_ssl_password` | `""` | Optional password for the client private key. | | `kafka_max_message_size` | `1048576` (1 MiB) | The largest transmittable message size in bytes, applied to producers (before compression) and consumers alike. | `KafkaConfig` carries further parameters that only concern one side of the event flow; they are documented where they apply: [compression and correlation ID generation](publisher.qmd#configuration) on the publisher page, and [retries and the dead letter queue](subscriber.qmd#configuration) on the subscriber page. ### Providing Configuration Values Since `KafkaConfig` is a [pydantic-settings](https://docs.pydantic.dev/latest/concepts/pydantic_settings/) `BaseSettings` class, every parameter can be supplied as an environment variable named after the field (case-insensitive; list values as JSON): ```bash export SERVICE_NAME="user-registry" export SERVICE_INSTANCE_ID="instance-001" export KAFKA_SERVERS='["kafka-server-1:9092", "kafka-server-2:9092"]' ``` In a typical service, `KafkaConfig` is not instantiated on its own. Instead, the service defines one config class inheriting from all the hexkit config classes it needs (plus its own settings) and decorates it with hexkit's `config_from_yaml`, which layers the configuration sources — keyword arguments override environment variables, which override secret files under `/secrets`, which override a config YAML, which overrides the defaults: ```python from hexkit.config import config_from_yaml from hexkit.providers.akafka import KafkaConfig @config_from_yaml(prefix="my_service") class Config(KafkaConfig): """Configuration for my service.""" ... # further settings, e.g. from other hexkit config classes config = Config() # reads .my_service.yaml, MY_SERVICE_* env vars, etc. ``` With a prefix set, environment variables take the form `MY_SERVICE_KAFKA_SERVERS='["localhost:9092"]'`, and the config YAML is looked up via the `MY_SERVICE_CONFIG_YAML` environment variable or as `.my_service.yaml` in the current or home directory. ## Kafka Test Utils The `hexkit.providers.akafka.testutils` module provides pytest fixtures that spin up a disposable Kafka broker in a container, so integration tests exercise real broker semantics without external infrastructure — publish test events, record what a service publishes, and assert expectations. The [Kafka Test Utils](test_utils.qmd) page documents the fixtures, the `KafkaFixture` utility methods, and common testing patterns, including in-memory alternatives for broker-less unit tests. ### Kafka Event Publisher The [`KafkaEventPublisher`](../../../reference/providers.akafka.KafkaEventPublisher.qmd) is hexkit's provider for publishing events to Apache Kafka. It implements the [`EventPublisherProtocol`](../../../reference/protocols.eventpub.EventPublisherProtocol.qmd), so application code programs against the protocol and stays independent of the event-streaming backend. In line with the [triple hexagonal architecture](../../arch_concepts/triple_hexagonal_arch.qmd), services usually do not call the provider directly from their core. Instead, an outbound *translator* offers the core a domain-bound API (say, `publish_user_created(user)`) and internally maps it to a `publish()` call with the right topic, type, and payload — see the [producers section](../../arch_concepts/event_driven_arch.qmd#producers) of the Event-Driven Architecture chapter. This page describes the provider itself; the examples call it directly for brevity. ## Configuration The publisher is configured with the shared [`KafkaConfig`](index.qmd#configuration) class. Beyond the connection parameters, three of its fields concern publishing specifically: | Parameter | Default | Description | | --------- | ------- | ----------- | | `generate_correlation_id` | `True` | If a publish is attempted without a correlation ID set in the context, generate a new one instead of raising an error. See [below](#headers-and-correlation-ids). | | `kafka_max_message_size` | `1048576` (1 MiB) | The largest transmittable message size in bytes, measured *before* compression. When compression is enabled, this may exceed the broker's `message.max.bytes`, which limits the compressed size. | | `kafka_compression_type` | `None` | Compression applied by the producer: `None`, `"gzip"`, `"snappy"`, `"lz4"`, or `"zstd"`. If unsure, `zstd` offers a good balance between speed and compression ratio. Consumers decompress transparently, whatever the setting. | ## Creating a Publisher The provider follows hexkit's async-context-manager construction pattern: the context manager starts the underlying `AIOKafkaProducer` on entry and flushes and stops it on exit. ```python from hexkit.providers.akafka import KafkaConfig, KafkaEventPublisher config = KafkaConfig( service_name="user-registry", service_instance_id="instance-001", kafka_servers=["localhost:9092"], ) async with KafkaEventPublisher.construct(config=config) as publisher: ... # use the publisher for the lifetime of the service ``` The publisher is typically constructed once at service startup and shared for the whole service lifetime — not opened per event. ## Publishing Events The single operation is `publish()`, with the event's four essential ingredients as required keyword arguments: ```python await publisher.publish( payload={"user_id": "abc123", "name": "Jane", "registered": "2026-07-01"}, type_="user_created", key="abc123", topic="users", ) ``` - `payload`: the event's content as a JSON-serializable dictionary. - `type_`: the event type, allowing consumers to distinguish different kinds of events within one topic. Transmitted as an event header. - `key`: the event key. Kafka guarantees ordering per key within a partition, so choose the key such that events that must stay in order share it — typically the ID of the affected resource. - `topic`: the topic to publish to. Topic, type, and key must be ASCII-only strings; they are validated before anything is sent. The payload is serialized as JSON with a few conveniences on top of the standard types: `UUID` and `Path` values are serialized as strings, and `date`/`datetime` values as ISO-8601 strings. The call returns once the broker has acknowledged the event, so a returned `publish()` means the event has been durably received by Kafka. Two optional arguments, `event_id` and `headers`, are covered in the next two sections. ## Event IDs Every event carries a UUID (version 4) identifying it, transmitted in the `event_id` header. If you do not pass one, `publish()` generates a fresh one — for ordinary publishing you never think about event IDs. Passing an explicit `event_id` is for cases where the ID must be controlled from the outside, chiefly when re-emitting an event that already has an identity. The [`PersistentKafkaPublisher`](../mongokafka.qmd#persistent-publisher), for example, stores each event's ID and reuses it on republishing, so consumers can recognize the event as the same one. ## Headers and Correlation IDs Along with the payload, each event carries metadata headers. Three header names are *reserved* and always set by the provider: - `type`: the event type passed as `type_`. - `event_id`: the event ID, as described above. - `correlation_id`: the ID tracing the request flow this event belongs to. Supplying one of the reserved names yourself in `headers` does not fail, but the value is overwritten and a warning is logged. Any other entries in the `headers` mapping are transmitted as additional headers; keys and values must be ASCII-encodable strings. The correlation ID is taken from the current async context, where it was set by whatever triggered this code path (for example, an inbound REST request or the [event subscriber](subscriber.qmd) consuming an upstream event). This is how hexkit traces request flows across service boundaries; the [Correlation IDs](../../observability_tools/correlation_ids.qmd) chapter explains the mechanism. If no correlation ID is set in the context, the behavior depends on the `generate_correlation_id` setting: - `True` (the default): a new correlation ID is generated and used, with an informational log message. Convenient for flows that legitimately originate here. - `False`: the publish fails with a `CorrelationIdContextError`. Use this strict mode to catch code paths that forgot to propagate the correlation ID. ## Error Handling `publish()` validates its inputs before contacting the broker: - a non-ASCII `topic`, `type_`, or `key` raises a [`NonAsciiStrError`](../../../reference/utils.NonAsciiStrError.qmd); - an `event_id` that is not a `UUID` raises a `TypeError`; - a payload containing values that are not JSON-serializable (after the conveniences noted above) raises a `TypeError`. A missing correlation ID raises a [`CorrelationIdContextError`](../../../reference/correlation.CorrelationIdContextError.qmd) when `generate_correlation_id` is disabled. Failures at the broker level — connection loss, an event exceeding `kafka_max_message_size`, or the broker rejecting the request — surface as exceptions from the underlying aiokafka producer (subclasses of `aiokafka.errors.KafkaError`). Since `publish()` waits for the broker's acknowledgment, no error means the event is safely stored in Kafka; there is no silent fire-and-forget failure mode. Note that a `KafkaEventPublisher` publishes ephemerally: once the call returns, the service holds no record of the event, and there is no built-in way to publish the same event again. If events must be re-emittable, use the [`PersistentKafkaPublisher`](../mongokafka.qmd#persistent-publisher) or, for domain-object state, the [outbox publisher](../mongokafka.qmd#outbox-publisher) — the [Publisher Variants](../../developer_help/outbox_v_persistent_pub.qmd) chapter compares the two. ## Testing The [Kafka test utils](test_utils.qmd) let you assert on published events against a real, containerized broker via `record_events()` and `expect_events()`. For runnable end-to-end examples, see the [Kafka pub/sub integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/eventpubsub/test_akafka_pubsub.py) in the hexkit repository. ### Kafka Event Subscriber The [`KafkaEventSubscriber`](../../../reference/providers.akafka.KafkaEventSubscriber.qmd) is hexkit's provider for consuming events from Apache Kafka. It handles the broker mechanics — subscriptions, deserialization, offsets, retries, and the dead letter queue — and hands each event to a service-defined *translator* that connects the event stream to the service's core logic. Unlike the [publisher](publisher.qmd), which can be used directly, the subscriber always requires a translator: consuming an event only makes sense in terms of what the service does with it. ## The Translator A translator implements the [`EventSubscriberProtocol`](../../../reference/protocols.eventsub.EventSubscriberProtocol.qmd) (from `hexkit.protocols.eventsub`), which requires: - `topics_of_interest`: the topics the provider shall subscribe to. - `types_of_interest`: the event types to act on. Events of other types are filtered out by the provider and never reach the translator. - `_consume_validated()`: the processing logic. It receives the event's `payload`, `type_`, `topic`, `key`, and `event_id` (the public `consume()` method validates these and then delegates). A typical translator validates the payload against a schema and calls a method on one of the service's core ports: ```python from pydantic import UUID4, BaseModel, Field, ValidationError from pydantic_settings import BaseSettings from hexkit.custom_types import Ascii, JsonObject from hexkit.protocols.eventsub import EventSubscriberProtocol class UserEventsConfig(BaseSettings): """Topic and type settings, kept configurable like all contractual details.""" users_topic: str = Field(default=..., examples=["users"]) class UserCreatedPayload(BaseModel): """Expected payload schema for user_created events.""" user_id: str name: str class UserEventsTranslator(EventSubscriberProtocol): """Consumes user events and forwards them to the core.""" def __init__(self, *, config: UserEventsConfig, registry: UserRegistryPort): """Takes the config and an inbound port of the service's core.""" self.topics_of_interest = [config.users_topic] self.types_of_interest = ["user_created"] self._registry = registry async def _consume_validated( self, *, payload: JsonObject, type_: Ascii, topic: Ascii, key: Ascii, event_id: UUID4, ) -> None: try: user = UserCreatedPayload.model_validate(payload) except ValidationError as error: raise self.MalformedPayloadError( f"Invalid user_created payload: {error}" ) from error await self._registry.register_user(user_id=user.user_id, name=user.name) ``` Raising an exception signals that consumption failed and triggers the [failure handling](#failure-handling-retries-and-the-dlq) described below; returning normally signals success. The protocol provides the `MalformedPayloadError` class (as an attribute of the translator base class) for payloads that do not match the expected schema. Since consumption is [at-least-once](#offset-management), translators must be *idempotent*: processing the same event twice must leave the service in the same state as processing it once. ### Combining Multiple Translators One `KafkaEventSubscriber` takes exactly one translator. To consume several concerns through a single consumer, bundle translators with the [`ComboTranslator`](../../../reference/providers.akafka.ComboTranslator.qmd): it merges their topics of interest and routes each incoming event to the right translator based on the event's topic and type. Two translators claiming the same topic/type combination raise a `ValueError` at construction time. A `ComboTranslator` may also include translators implementing the [`DaoSubscriberProtocol`](../../protocols/daosubscriber.qmd), so ordinary event consumption and [outbox consumption](../../protocols/daosubscriber.qmd) can share one consumer. ## Configuration The subscriber is configured with the shared [`KafkaConfig`](index.qmd#configuration) class. Two of the connection parameters play a special role on the consuming side: - `service_name` doubles as the *consumer group ID*. All instances of a service form one consumer group, so each event is processed by only one instance, and the workload is shared. - `kafka_max_message_size` bounds the largest fetchable message. New consumer groups start reading at the earliest available offset, so a freshly deployed service works through the retained event history rather than only seeing new events. The remaining consumer-side parameters control failure handling: | Parameter | Default | Description | | --------- | ------- | ----------- | | `kafka_max_retries` | `0` | How many times to immediately retry a failed event before giving up. | | `kafka_retry_backoff` | `0` | Seconds to wait before the first retry; the wait doubles with each subsequent retry. | | `kafka_enable_dlq` | `False` | Whether to publish events to the dead letter queue after retries are exhausted, instead of crashing. | | `kafka_dlq_topic` | `"dlq"` | The topic collecting failed events for inspection. | ## Creating and Running a Subscriber Like all hexkit providers, the subscriber is constructed through an async context manager. Passing a `dlq_publisher` — usually a [`KafkaEventPublisher`](publisher.qmd) — is required when the DLQ is enabled: ```python from hexkit.providers.akafka import ( KafkaConfig, KafkaEventPublisher, KafkaEventSubscriber, ) config = KafkaConfig( service_name="user-registry", service_instance_id="instance-001", kafka_servers=["localhost:9092"], kafka_enable_dlq=True, ) translator = UserEventsTranslator(config=user_events_config, registry=registry) async with ( KafkaEventPublisher.construct(config=config) as dlq_publisher, KafkaEventSubscriber.construct( config=config, translator=translator, dlq_publisher=dlq_publisher ) as subscriber, ): await subscriber.run() ``` `run()` blocks and consumes indefinitely — it is the main loop of an event-driven service. For tests and one-shot scenarios, `run(forever=False)` consumes exactly one event and returns. The provider subscribes to the translator's `topics_of_interest`. Incoming events are screened before they reach the translator: events whose type is not among the `types_of_interest`, or which lack a type or correlation ID entirely, are ignored (with a log message) and acknowledged. For each accepted event, the provider extracts the correlation ID from the event headers and sets it as a context variable, so all downstream processing — including any events the service publishes in turn — carries the same correlation ID automatically. ## Offset Management Kafka consumers track their progress through *offsets*. The subscriber disables auto-committing and commits offsets explicitly — only after an event has been fully handled, meaning the translator returned successfully, the event was ignored as not-of-interest, or the event was handed off to the DLQ. The consequence is an *at-least-once* guarantee: an event is never lost by being skipped, but if the service crashes mid-processing, the uncommitted event is redelivered after a restart — possibly to another instance in the consumer group. Together with event republishing and the [DLQ](#failure-handling-retries-and-the-dlq), this is why translators must be idempotent. ## Failure Handling: Retries and the DLQ Event consumers are long-lived, and a single poison event should not take down the whole service. When the translator raises an exception, the subscriber escalates in stages: 1. **Immediate retries.** If `kafka_max_retries` is greater than zero, the event is re-attempted up to that many times, waiting `kafka_retry_backoff` seconds before the first retry and doubling the wait each time (a backoff of 2 with three retries waits 2, 4, then 8 seconds). 2. **Dead letter queue.** If retries are exhausted (or none are configured) and `kafka_enable_dlq` is `True`, the event is published to the `kafka_dlq_topic` via the provided `dlq_publisher`, and consumption continues with the next event. The DLQ event keeps the original payload, type, and key, gets a new event ID, and carries additional headers describing the failure: the exception class and message, the original topic, the original event ID, and the name of the service that failed. 3. **Crash.** With the DLQ disabled, the exception is re-raised after the retries and the service goes down without committing the offset — the event will be redelivered on restart. This is a reasonable mode for deployments where a supervisor restarts the service and event order matters more than liveness. Failed events that were parked in the DLQ can be inspected and, once the cause is resolved, republished to a service-specific retry topic (`retry-`). The subscriber automatically subscribes to its retry topic whenever the DLQ is enabled and restores each retried event's original topic before passing it to the translator, so translators never need to know whether an event arrived directly or via the DLQ detour. The full DLQ lifecycle — inspecting, resolving, and requeueing events, and building DLQ-processing services with the [`DLQSubscriberProtocol`](../../../reference/protocols.eventsub.DLQSubscriberProtocol.qmd) — is described in the [Dead Letter Queue](../../arch_concepts/dlq.qmd) chapter. ## Testing The [Kafka test utils](test_utils.qmd) provide a containerized broker for integration-testing translators through a real subscriber; `run(forever=False)` consumes a single published test event. For runnable examples, see the [Kafka pub/sub integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/eventpubsub/test_akafka_pubsub.py) and the [DLQ integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/eventpubsub/test_akafka_dlqsub.py) in the hexkit repository. ### Kafka Test Utils The `hexkit.providers.akafka.testutils` module provides pytest fixtures and helpers for testing event-driven services against a real, disposable Kafka broker. The broker runs in a container managed by [testcontainers](https://testcontainers-python.readthedocs.io/), so integration tests exercise actual broker semantics — topics, partitions, offsets, retention — without any external infrastructure. All that is required is a working Docker (or compatible) daemon. Install hexkit with the `test-akafka` extra to use these utilities (it adds testcontainers and the pytest plugins on top of the `akafka` extra): ```bash pip install hexkit[test-akafka] ``` For unit tests that should not use a live broker, hexkit also ships an [in-memory event publisher](#unit-testing-without-a-broker), described at the end of this page. ## The Fixtures Two fixtures work in tandem, splitting the work between pytest scopes: - `kafka_container_fixture` (fixture name `kafka_container`) is *session-scoped*: it starts one Kafka test container (`confluentinc/cp-kafka`) per test session, which all tests reuse. Starting the container takes a few seconds, so sharing it keeps the test suite fast. - `kafka_fixture` (fixture name `kafka`) is *function-scoped* and async: it yields a [`KafkaFixture`](#the-kafkafixture) bound to that container and clears all topics before each test, so every test starts from an empty broker. `clean_kafka_fixture` is an alias for it. - `persistent_kafka_fixture` (also named `kafka`) is the variant *without* the upfront cleanup, for tests that intentionally share broker state — for example, when asserting on events retained in compacted topics across test phases. Both the container fixture and the `KafkaFixture`-producing fixture must be imported into the test module (or a `conftest.py`) for pytest to resolve them, hence the `# noqa: F401` comments: ```python import pytest from hexkit.providers.akafka.testutils import ( KafkaFixture, kafka_container_fixture, # noqa: F401 kafka_fixture, # noqa: F401 ) @pytest.mark.asyncio async def test_something(kafka: KafkaFixture): ... ``` If the default scopes or names do not fit your test setup — say, you want a module-scoped fixture, or both a clean and a persistent fixture side by side (they otherwise collide on the name `kafka`) — the factory functions `get_kafka_container_fixture`, `get_clean_kafka_fixture`, and `get_persistent_kafka_fixture` create the same fixtures with custom `scope` and `name` arguments: ```python from hexkit.providers.akafka.testutils import get_clean_kafka_fixture module_kafka_fixture = get_clean_kafka_fixture(scope="module", name="module_kafka") ``` ## The KafkaFixture A `KafkaFixture` carries three attributes: - `config`: a ready-made [`KafkaConfig`](../../../reference/providers.akafka.KafkaConfig.qmd) pointing at the containerized broker — pass it to the providers under test. - `kafka_servers`: the bootstrap server list, for constructing your own config (e.g. to add [DLQ settings](subscriber.qmd#configuration)). - `publisher`: a started [`KafkaEventPublisher`](../../../reference/providers.akafka.KafkaEventPublisher.qmd) connected to the broker. On top of that, it offers utility methods for the three things Kafka tests do all the time: feeding events in, asserting on events coming out, and resetting broker state. ### Publishing Test Events `publish_event()` publishes a single event through the fixture's built-in publisher — typically to feed an inbound event to a subscriber under test: ```python await kafka.publish_event( payload={"user_id": "abc123", "name": "Jane"}, type_="user_created", topic="users", key="abc123", # defaults to "test" if omitted ) ``` `event_id` and `headers` can be passed as well; by default a fresh event ID and correlation ID are generated, just as with a production publisher. ### Asserting on Published Events `expect_events()` returns an async context manager that records everything published to a topic while the `async with` block runs, and on exit checks the recorded events against a sequence of `ExpectedEvent`s — raising a `ValidationError` (with a diff-friendly message) on any mismatch: ```python from hexkit.providers.akafka.testutils import ExpectedEvent async with kafka.expect_events( events=[ExpectedEvent(payload={"user_id": "abc123"}, type_="greeting_sent")], in_topic="greetings", ): ... # run the service logic that should publish exactly this event ``` The check is strict about count and order: the recorded events must match the expected ones one-to-one, in sequence. Per event, the following is compared: - `payload` and `type_` are always compared. The expected payload may contain the same rich values a publisher accepts (UUIDs, datetimes, ...) — it is normalized with the publisher's JSON serializer before comparison. - `key`, `event_id`, and `headers` are optional and only compared when set to a value other than `None`. Events are recorded without their headers by default. To assert on headers, pass `capture_headers=True` — expecting headers without it raises a `ValidationError`. (The `type` and `event_id` headers are extracted into their own fields and are not part of the compared headers.) ### Recording Events for Inspection When an exact expectation is too rigid — the number of events varies, or you want to assert on selected fields only — `record_events()` provides the same recording without the automatic check. It returns an `EventRecorder` whose `recorded_events` (a sequence of `RecordedEvent`s with `payload`, `type_`, `key`, `event_id`, and optionally `headers` fields) are available after the context block exits: ```python async with kafka.record_events(in_topic="notifications") as recorder: ... # run the service logic assert len(recorder.recorded_events) == 2 assert recorder.recorded_events[0].type_ == "email_sent" ``` Both `expect_events()` and `record_events()` only consider events published *inside* the context block: on entry, the recorder notes the topic's current offsets, and on exit it consumes exactly the events published since then. This also means the recorder never waits indefinitely for events that were expected but not published — it simply reports the mismatch. If you need to compare manually recorded events against expectations later, the module-level `check_recorded_events()` function performs the same validation that `expect_events()` runs on exit. ### Managing Broker State - `clear_topics(topics=..., exclude_internal=True)` deletes all events from the given topic(s), or from all topics when none are specified (internal `__`-prefixed topics are excluded by default). The `kafka_fixture` calls this before every test; call it yourself to separate phases within one test or when using the persistent fixture. Compacted topics are handled by temporarily switching their cleanup policy. - `set_cleanup_policy(topic=..., policy=...)` and `get_cleanup_policy(topic=...)` set or read a topic's `cleanup.policy` — the way to test behavior on compacted topics (see the [persistent publisher](../mongokafka.qmd#persistent-publisher)), e.g. `policy="compact"`. - `set_topic_config(topic=..., config={...})` and `get_topic_config(topic=..., config_name=...)` do the same for arbitrary topic configuration values. ## Common Testing Patterns ### Testing the Publishing Side Wrap the service logic in `expect_events()` and let it assert that the right events — and only those — were published: ```python @pytest.mark.asyncio async def test_greeting_published(kafka: KafkaFixture): async with KafkaEventPublisher.construct(config=kafka.config) as publisher: emitter = GreetingEmitter(config=greeting_config, publisher=publisher) async with kafka.expect_events( events=[ ExpectedEvent( payload={"user_id": "abc123", "text": "Hello, Jane!"}, type_="greeting_sent", key="abc123", ) ], in_topic="greetings", ): await emitter.greet(user_id="abc123", name="Jane") ``` ### Testing the Consuming Side Publish an inbound event with `publish_event()`, then run a real [`KafkaEventSubscriber`](subscriber.qmd) with `run(forever=False)`, which consumes exactly one event and returns. Note that because the consumer waits for an event even if `forever=False`, your tests might hang if no event is published to the specified topic. Assert on the side effects in the service's core: ```python from hexkit.providers.akafka import KafkaEventSubscriber @pytest.mark.asyncio async def test_user_created_consumed(kafka: KafkaFixture): await kafka.publish_event( payload={"user_id": "abc123", "name": "Jane"}, type_="user_created", topic="users", key="abc123", ) registry = InMemUserRegistry() # test double for the core port translator = UserEventsTranslator(config=user_events_config, registry=registry) async with KafkaEventSubscriber.construct( config=kafka.config, translator=translator ) as subscriber: await subscriber.run(forever=False) assert registry.has_user("abc123") ``` ### Mocking the Translator To test event *routing* — that the subscriber picks up the right topics and delivers events with the right arguments — a mock can stand in for the translator. Any object with `topics_of_interest`, `types_of_interest`, and an async `consume()` method satisfies the provider, so an `AsyncMock` works directly: ```python from unittest.mock import AsyncMock @pytest.mark.asyncio async def test_event_reaches_translator(kafka: KafkaFixture): translator = AsyncMock() translator.topics_of_interest = ["users"] translator.types_of_interest = ["user_created"] event_id = uuid4() await kafka.publish_event( payload={"user_id": "abc123"}, type_="user_created", topic="users", key="abc123", event_id=event_id, ) async with KafkaEventSubscriber.construct( config=kafka.config, translator=translator ) as subscriber: await subscriber.run(forever=False) translator.consume.assert_awaited_once_with( payload={"user_id": "abc123"}, type_="user_created", topic="users", key="abc123", event_id=event_id, ) ``` For testing the translator's *logic*, no broker is needed at all — see [below](#unit-testing-without-a-broker). ### Testing DLQ Behavior Failure handling needs a config with the [DLQ enabled](subscriber.qmd#failure-handling-retries-and-the-dlq); derive one from the fixture's connection details and record what ends up in the DLQ topic: ```python config = KafkaConfig( service_name="user-registry", service_instance_id="test", kafka_servers=kafka.kafka_servers, kafka_enable_dlq=True, ) ``` For complete examples, see the [DLQ integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/eventpubsub/test_akafka_dlqsub.py) in the hexkit repository, alongside the [pub/sub integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/eventpubsub/test_akafka_pubsub.py) that exercise all of the utilities described on this page. ## Testing SSL Connections The default test container speaks plaintext. To test TLS-secured connections — certificate validation, client authentication, the `kafka_ssl_*` [configuration parameters](index.qmd#configuration) — the `hexkit.providers.akafka.testcontainer` module provides `KafkaSSLContainer`, a Kafka test container configured with PEM certificates: ```python from hexkit.providers.akafka.testcontainer import KafkaSSLContainer with KafkaSSLContainer( cert=broker_cert, # PEM certificate (chain) of the broker key=broker_key, # PEM private key of the broker password=broker_key_password, trusted=ca_cert, # trusted CA certificates client_auth="required", # or "requested" or "none" ) as kafka_container: kafka_servers = [kafka_container.get_bootstrap_server()] ... # build a KafkaConfig with kafka_security_protocol="SSL" ``` This container is not wrapped in a fixture; manage it directly in tests that need it. ## Unit Testing Without a Broker Spinning up a container is worth it for integration tests, but plain unit tests are better served without one: - **Publishing side:** the [`InMemEventPublisher`](../../../reference/providers.testing.InMemEventPublisher.qmd) (from [`hexkit.providers.testing`](../testing.qmd)) implements the [`EventPublisherProtocol`](../../../reference/protocols.eventpub.EventPublisherProtocol.qmd) against an in-memory [`InMemEventStore`](../../../reference/providers.testing.InMemEventStore.qmd) that keeps one queue per topic. Inject it into the code under test in place of a `KafkaEventPublisher` and pop the published events off the store: ```python from hexkit.providers.testing import InMemEventPublisher publisher = InMemEventPublisher() emitter = GreetingEmitter(config=greeting_config, publisher=publisher) await emitter.greet(user_id="abc123", name="Jane") event = publisher.event_store.get("greetings") # raises TopicExhaustedError if empty assert event.type_ == "greeting_sent" assert event.payload == {"user_id": "abc123", "text": "Hello, Jane!"} ``` Reading from an empty topic raises a [`TopicExhaustedError`](../../../reference/providers.testing.TopicExhaustedError.qmd) — handy for asserting that *nothing* was published. Note that the store is a simple same-thread queue: it is not suitable for inter-thread or inter-process communication, and there is no in-memory subscriber consuming from it. - **Consuming side:** a translator is an ordinary object — call its `consume()` method directly with a test payload and assert on the effects, no subscriber or broker involved: ```python translator = UserEventsTranslator(config=user_events_config, registry=registry) await translator.consume( payload={"user_id": "abc123", "name": "Jane"}, type_="user_created", topic="users", key="abc123", event_id=uuid4(), ) assert registry.has_user("abc123") ``` These in-memory approaches verify your logic; they do not verify serialization, topic wiring, offset handling, or retry behavior. Cover those with the container-based fixtures above. ### MongoDB ## MongoDB DAO The MongoDB DAO provider implements the [DAO protocol](../protocols/dao.qmd) on top of MongoDB, storing each resource as a document in a collection. It consists of [`MongoDbDaoFactory`](../../reference/providers.mongodb.MongoDbDaoFactory.qmd), which implements the `DaoFactoryProtocol`, and [`MongoDbDao`](../../reference/providers.mongodb.MongoDbDao.qmd), the DAO objects it produces. This section covers everything MongoDB-specific; the backend-independent semantics of the DAO operations are documented in the [DAO protocol](../protocols/dao.qmd) chapter. To use it, install hexkit with the `mongodb` extra: ```sh pip install hexkit[mongodb] ``` ### Configuration Connection parameters are collected in [`MongoDbConfig`](../../reference/providers.mongodb.MongoDbConfig.qmd), a `pydantic_settings.BaseSettings` class. Inherit your service's config class from it, or instantiate it directly: ```python from hexkit.providers.mongodb import MongoDbConfig config = MongoDbConfig( mongo_dsn="mongodb://localhost:27017", db_name="library", ) ``` | Parameter | Description | | --------- | ----------- | | `mongo_dsn` | The MongoDB connection string, possibly including credentials. Stored as a Pydantic `Secret` so it does not leak into logs or error messages. | | `db_name` | The name of the database to use on the server. | | `mongo_timeout` | Optional timeout in seconds applied to every database operation, covering server selection, connection checkout, and execution. When exceeded, operations raise a `DbTimeoutError`. Defaults to `None` (no timeout). | ### Creating DAOs The recommended way to set up the factory is the `construct()` async context manager, which opens the MongoDB client on entry and closes it on exit: ```python from pydantic import BaseModel, ConfigDict, UUID4 from hexkit.protocols.dao import UUID4Field from hexkit.providers.mongodb import MongoDbDaoFactory class Book(BaseModel): """A book in our catalog.""" model_config = ConfigDict(frozen=True) id: UUID4 = UUID4Field(description="The ID of the book.") title: str author: str pages: int async with MongoDbDaoFactory.construct(config=config) as dao_factory: book_dao = await dao_factory.get_dao( name="books", dto_model=Book, id_field="id" ) # use book_dao here ``` The `name` passed to `get_dao()` is used directly as the MongoDB collection name, so each resource type lives in its own collection. Neither the database nor the collection needs to exist beforehand; MongoDB creates both on first write. If you prefer to manage the client yourself (for example to share one client across several components), instantiate the factory directly with a `pymongo` `AsyncMongoClient`. The `ConfiguredMongoClient` context manager creates a client with the same settings that `construct()` uses: ```python from hexkit.providers.mongodb import ConfiguredMongoClient async with ConfiguredMongoClient(config=config) as client: dao_factory = MongoDbDaoFactory(config=config, client=client) ``` ### How DTOs Map to Documents The provider converts DTOs to documents with Pydantic's `model_dump()` and renames the configured ID field to MongoDB's internal `_id` on the way in (and back on the way out). A `Book` with `id=UUID("18d90a54...")` is stored as: ```json { "_id": UUID("18d90a54-e79b-4d92-a844-e8fa9dfda2ac"), "title": "Refactoring", "author": "Martin Fowler", "pages": 448 } ``` Values keep their Python types during the dump, so they are stored using MongoDB's native representations where available: UUIDs become BSON UUIDs (the client is configured with the `standard` UUID representation) and datetimes become BSON dates, returned as timezone-aware `datetime` objects when read back. Since MongoDB stores datetimes with millisecond precision, truncate them accordingly on your DTOs (the `hexkit.utils.now_utc_ms_prec` helper creates compliant timestamps), otherwise a value may lose precision in the database and no longer compare equal after a roundtrip. Field types without a BSON equivalent (such as `pathlib.Path`) need a `field_serializer` on the DTO model that converts them to something storable; the provider raises an error otherwise. You never interact with `_id` yourself: in DTOs, find mappings, sort specifications, and index definitions, always use the field name from your model. ### CRUD Operations All seven operations of the [DAO protocol](../protocols/dao.qmd#crud-operations) are supported, with the semantics and exceptions described there: ```python book = Book(title="Refactoring", author="Martin Fowler", pages=448) # Create (fails with ResourceAlreadyExistsError if the ID is taken): await book_dao.insert(book) # Read (fails with ResourceNotFoundError if the ID is unknown): book = await book_dao.get_by_id(book.id) # Update (fails with ResourceNotFoundError if the ID is unknown): await book_dao.update(book.model_copy(update={"pages": 464})) # Upsert (update or insert, never fails on existence): await book_dao.upsert(book) # Delete (fails with ResourceNotFoundError if the ID is unknown): await book_dao.delete(book.id) ``` Under the hood, `update` and `upsert` replace the whole document; partial updates are not part of the DAO abstraction. ### Queries and Filters The find methods take a `mapping` of field names to expected values, and all entries must match (a logical AND). Pass values with the same Python types as on your model: a `UUID` for a UUID field, a timezone-aware `datetime` for a datetime field, and so on, since matching happens against the natively stored values: ```python # Exactly one hit expected (raises NoHitsFoundError / MultipleHitsFoundError): book = await book_dao.find_one(mapping={"title": "Refactoring"}) # Zero or more hits, consumed by async iteration: async for book in book_dao.find_all(mapping={"author": "Martin Fowler"}): print(book.title) ``` Mapping keys are validated against the DTO model, and unknown field names raise an `InvalidFindMappingError`. Two MongoDB-specific extensions pass through this validation: - **Dot notation** reaches into nested documents. Only the top-level field is validated against the model: `{"address.city": "Tübingen"}`. - **MongoDB query operators** can be used as values, giving you access to the [MongoDB query language](https://www.mongodb.com/docs/manual/reference/operator/query/) where simple equality is not enough: ```python # Range query: thick_books = book_dao.find_all(mapping={"pages": {"$gt": 300}}) # Membership test: selected = book_dao.find_all( mapping={"author": {"$in": ["Martin Fowler", "Kent Beck"]}} ) # Top-level logical operators work too: either = book_dao.find_all( mapping={"$or": [{"author": "Kent Beck"}, {"pages": {"$lt": 100}}]} ) ``` Keep in mind that operator-based mappings tie the calling code to MongoDB. That is a perfectly reasonable trade-off inside a translator, but if you want your code to stay provider-agnostic, stick to plain equality mappings. The [in-memory mock DAO](testing.qmd) understands a subset of these operators, so most operator-based queries remain testable without a database. For pagination and sorting, `find_all` accepts `skip`, `limit`, and `sort` arguments, and returns a [`FindResult`](../../reference/protocols.dao.FindResult.qmd) that also reports the total number of matches: ```python result = book_dao.find_all( mapping={"author": "Martin Fowler"}, skip=20, limit=10, sort=["title", "-pages"], # title ascending, then pages descending ) page = await result.to_list() total = await result.total_count() # all matches, ignoring skip and limit ``` The query is executed lazily as a single `find` with a cursor, so iterating a `FindResult` streams documents rather than loading everything into memory; `total_count()` triggers one additional count query on first call. ### Defining Indexes The provider always keeps a unique index on the ID field (MongoDB does this for `_id` automatically). Additional indexes are declared with [`MongoDbIndex`](../../reference/index.qmd#provider-mongodb) objects passed to `get_dao()`, which creates them on the collection when the DAO is constructed (creating an index that already exists is a no-op): ```python from hexkit.providers.mongodb import MongoDbIndex book_dao = await dao_factory.get_dao( name="books", dto_model=Book, id_field="id", indexes=[ # A simple ascending index over one field: MongoDbIndex(fields="author"), # A compound, unique index (1 = ascending, -1 = descending): MongoDbIndex( fields={"author": 1, "title": -1}, properties={"unique": True}, ), ], ) ``` The `fields` argument takes a single field name (indexed in ascending order) or a dictionary mapping field names to sort orders. All field names must exist on the DTO model, otherwise `get_dao()` raises an `IndexFieldsInvalidError`; if an index covers the ID field, use the model's field name and the provider translates it to `_id`. The `properties` dictionary is forwarded to pymongo's [`create_index()`](https://pymongo.readthedocs.io/en/stable/api/pymongo/collection.html#pymongo.collection.Collection.create_index), so all index properties supported there can be used, for example `unique`, `sparse`, or `expireAfterSeconds` for TTL indexes. A unique index is also the mechanism behind `UniqueConstraintViolationError`: writes that would break a unique index over non-ID fields raise it, letting you enforce uniqueness of, say, an email address at the database level. ### Error Handling The provider translates pymongo's errors into the exception types of the [DAO protocol](../protocols/dao.qmd#error-handling), so calling code never needs to import from pymongo: | Situation | Raised exception | | --------- | ---------------- | | `get_by_id`, `update`, or `delete` on a missing ID | `ResourceNotFoundError` | | `insert` with an already existing ID | `ResourceAlreadyExistsError` | | Write breaking a unique index over non-ID fields | `UniqueConstraintViolationError` | | Find mapping with unknown field names | `InvalidFindMappingError` | | `find_one` matching zero / more than one document | `NoHitsFoundError` / `MultipleHitsFoundError` | | Operation exceeding the configured `mongo_timeout` | `DbTimeoutError` | | Any other driver-level failure | `DaoError` | All of these derive from `DaoError`, so `except DaoError` is a safe catch-all for database problems. ### Transactions The `with_transaction()` method declared by the DAO protocol is not yet implemented by the MongoDB provider and raises `NotImplementedError`. Every single DAO operation is atomic on its own (MongoDB guarantees single-document atomicity), but grouping several operations into one transaction is not currently possible through the DAO interface. ### Testing For integration tests, the `test-mongodb` extra provides pytest fixtures based on [testcontainers](https://testcontainers-python.readthedocs.io) that spin up a disposable MongoDB instance. The `mongodb_fixture` yields a `MongoDbFixture` whose `dao_factory` attribute is a ready-to-use `MongoDbDaoFactory`: ```python from hexkit.providers.mongodb.testutils import ( MongoDbFixture, mongodb_container_fixture, # noqa: F401 mongodb_fixture, # noqa: F401 ) async def test_book_dao(mongodb: MongoDbFixture): book_dao = await mongodb.dao_factory.get_dao( name="books", dto_model=Book, id_field="id" ) ... ``` For unit tests that should not touch a database at all, the [in-memory mock DAO](testing.qmd) mimics the MongoDB provider's behavior, including a subset of the query operators. See also the [MongoDB Test Utils](#mongodb-test-utils) section below. The provider's own [integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/dao/test_mongodb_dao.py) double as an extensive collection of runnable usage examples. ## MongoDB Migration Tools When a service's data model evolves, documents already stored in MongoDB must be brought into the shape its DTO models now expect. The `hexkit.providers.mongodb.migrations` subpackage provides a versioned migration framework for exactly that, run by the service itself at startup. It is covered in the dedicated [MongoDB Migrations](mongodb_migrations.qmd) chapter. ## MongoDB KV Store Providers MongoDB providers implement `KeyValueStoreProtocol` with document-oriented, durable persistence. Available variants are `MongoDbBytesKeyValueStore`, `MongoDbStrKeyValueStore`, `MongoDbJsonKeyValueStore`, and `MongoDbDtoKeyValueStore`. API reference: [hexkit.providers.mongodb](../../reference/index.qmd#provider-mongodb) ## MongoDB Test Utils ### MongoDB Migrations MongoDB itself is schemaless, but the services built with hexkit are not: the [DAO pattern](../arch_concepts/dao_pattern.qmd) ties every collection to a Pydantic DTO model, and documents written by an older release of a service may no longer match the models of a newer one. The `hexkit.providers.mongodb.migrations` subpackage provides a versioned migration framework that closes this gap: it tracks a version number for the database, runs the migration code needed to reach the version a service release expects, and coordinates multiple service instances so that migrations run exactly once. The tooling ships with the regular MongoDB provider extra: ```sh pip install hexkit[mongodb] ``` Working with it involves three pieces: - a [`MigrationDefinition`](../../reference/providers.mongodb.migrations.MigrationDefinition.qmd) subclass per database version, containing the actual transformation logic, - the [`MigrationManager`](../../reference/providers.mongodb.migrations.MigrationManager.qmd), which every service instance runs at startup to apply pending migrations or wait for another instance to finish them, - [`MigrationConfig`](../../reference/providers.mongodb.migrations.MigrationConfig.qmd), the configuration class tying it all together. ## How Versioning Works The framework stores its bookkeeping in a dedicated collection of the service's database, named by the `db_version_collection` config option. Every completed migration run inserts a record there with the reached `version`, a `completed` timestamp, whether the run was `backward`, and its duration. The current database version is the `version` of the most recently completed record — not simply the highest number, so rollbacks (see [Reversible Migrations](#reversible-migrations)) are handled correctly. **Version 1 is reserved by the framework**: it marks the point where versioning was initialized. On the very first run against an unversioned database, the manager creates the version collection and records version 1 without running any migration code. The first migration you write yourself is therefore always version 2. The same collection also holds a lock document while a migration is in progress, which is how concurrently starting service instances avoid migrating twice. ## Configuration [`MigrationConfig`](../../reference/providers.mongodb.migrations.MigrationConfig.qmd) extends [`MongoDbConfig`](../../reference/providers.mongodb.MongoDbConfig.qmd) (so it carries `mongo_dsn`, `db_name`, and `mongo_timeout` as described in the [MongoDB chapter](mongodb.qmd#configuration)) with three migration-specific settings: | Parameter | Description | | --------- | ----------- | | `db_version_collection` | Name of the collection holding the DB version records and the migration lock, e.g. `"ifrsDbVersions"`. | | `migration_wait_sec` | How many seconds an instance waits between checks while another instance is migrating. | | `migration_max_wait_sec` | Upper limit in seconds on that waiting before a `MigrationTimeoutError` is raised. Defaults to `None`, meaning wait indefinitely. | Inherit your service's config class from `MigrationConfig` instead of `MongoDbConfig` to make the additional settings part of the service configuration. ## Defining Migrations Each database version is described by a subclass of [`MigrationDefinition`](../../reference/providers.mongodb.migrations.MigrationDefinition.qmd) that assigns the `version` class attribute and implements the async `apply()` method. `apply()` contains whatever it takes to move the database *from the previous version to `version`*: ```python from hexkit.providers.mongodb.migrations import Document, MigrationDefinition from my_service.models import User # current DTO model of the users collection class V2Migration(MigrationDefinition): """Add the 'is_active' flag introduced in v2 of the service.""" version = 2 async def apply(self): async def add_active_flag(doc: Document) -> Document: doc["is_active"] = True return doc async with self.auto_finalize(coll_names=["users"], copy_indexes=True): await self.migrate_docs_in_collection( coll_name="users", change_function=add_active_flag, validation_model=User, id_field="user_id", ) ``` Within `apply()`, the base class provides the building blocks described below. `self._db` holds the pymongo `AsyncDatabase`, so operations beyond these helpers (creating new collections, dropping obsolete ones, etc.) are always possible. ### Transforming Documents `migrate_docs_in_collection()` is the workhorse: it reads every document of a collection, passes it through the given async *change function* (old document in, new document out), and writes the results to a *temporary* collection (e.g. `tmp_v2_new_users`). The original collection remains untouched at this point. Documents are read and inserted in batches; the `batch_size` argument (default 1000) controls both the cursor batch size and the size of the bulk inserts, which is worth tuning for very large documents. Since change functions operate on raw documents, remember that the document contains MongoDB's `_id` field rather than the DTO model's ID field name, and values appear as stored in BSON (see [How DTOs Map to Documents](mongodb.qmd#how-dtos-map-to-documents)). ### Validating the Results Passing `validation_model` and `id_field` to `migrate_docs_in_collection()` cross-checks every migrated document against the given DTO model: the document must populate the model without errors and convert back to the identical document. This catches drift between migration code and model definitions early, instead of at the first read by the service. The check only runs when the migration is the *final* one of the run, because intermediate versions of a multi-step run generally do not match the models currently defined in code (those always describe the latest version). Pass `force_validate=True` to validate regardless, e.g. with a model you defined locally inside the migration. The underlying helper, [`validate_doc`](../../reference/providers.mongodb.migrations.validate_doc.qmd), can also be called directly for custom validation logic. ### Staging and Finalizing Because the transformed documents live in temporary collections, they must be *staged* once the transformation is complete: - `stage_new_collections(coll_names)` moves the original collections aside by renaming them (e.g. `users` → `tmp_v2_old_users`) and renames the temporary new collections to the original names. Staging a collection that does not exist is skipped with a warning, so migrations also run cleanly against an empty database. - `drop_old_collections(enforce_indexes=...)` afterwards drops the set-aside originals, making the migration final. Indexes are **not** carried over automatically when documents are rewritten into a new collection. `auto_copy_indexes(coll_names=...)` copies all indexes (including their options such as `unique`) from the old to the new collections; `copy_indexes()` does the same for a single explicitly named source/destination pair. Since dropping the old collections is the point of no return, `drop_old_collections(enforce_indexes=True)` refuses to proceed if indexes were not copied first — use it whenever the migration leaves the collections' indexes applicable; leave it `False` when the migration deliberately changes the index layout and you handle index creation yourself. Rather than calling these steps manually, most migrations should wrap the transformation in the `auto_finalize()` context manager, as shown in the example above. On a successful exit of the `async with` block it stages the listed collections, copies indexes if `copy_indexes=True`, and drops the old collections. If anything fails inside the block or during finalization, it unstages whatever was already staged and drops the temporary collections, so the database is left in its pre-migration state (should that cleanup itself fail, a critical log message recommends a database restore). ## Reversible Migrations Migrations are forward-only by default. To support downgrades — for example, to roll back a botched service release — additionally inherit from the [`Reversible`](../../reference/providers.mongodb.migrations.Reversible.qmd) mixin and implement `unapply()`, which must undo what `apply()` did, moving the database back *from `version` to the previous version*: ```python from hexkit.providers.mongodb.migrations import ( Document, MigrationDefinition, Reversible, ) class V2Migration(MigrationDefinition, Reversible): version = 2 async def apply(self): ... # as above async def unapply(self): async def drop_active_flag(doc: Document) -> Document: doc.pop("is_active", None) return doc async with self.auto_finalize(coll_names=["users"], copy_indexes=True): await self.migrate_docs_in_collection( coll_name="users", change_function=drop_active_flag, ) ``` All the building blocks above work the same way in `unapply()`; the temporary collections are prefixed with `tmp_v2_unapply_` instead of `tmp_v2_` to keep the two directions apart. When the manager finds the database at a *higher* version than the service expects, it runs the affected migrations' `unapply()` methods in descending order. If one of them is not reversible, the backward migration fails with an error — so whether to offer reversibility is decided per migration. ## Running Migrations at Startup Migrations are not run by a separate deployment step but by the service itself: every instance executes the [`MigrationManager`](../../reference/providers.mongodb.migrations.MigrationManager.qmd) *before* starting its actual work (consuming events, serving requests, etc.). The manager is an async context manager; `migrate_or_wait()` does the work: ```python from hexkit.providers.mongodb.migrations import MigrationManager from my_service.config import Config # inherits from MigrationConfig from my_service.migrations import V2Migration, V3Migration DB_VERSION = 3 # the database version this service release expects MIGRATION_MAP = {2: V2Migration, 3: V3Migration} async def run_db_migrations(): """Call this before the service begins its main execution loop.""" config = Config() async with MigrationManager(config, DB_VERSION, MIGRATION_MAP) as mm: await mm.migrate_or_wait() ``` The `migration_map` gathers all known migration definitions, keyed by the version each one migrates *to*. `target_version` is the version this release of the service needs. From those, `migrate_or_wait()` determines what to do: - **Database already at the target version**: return immediately; the service starts right away. This is the steady-state cost of the mechanism — a single read of the version collection. - **Database behind the target version**: acquire the migration lock and apply the missing migrations in ascending order (for a database at version 2 and a target of 5, that is versions 3, 4, and 5), then record the new version and release the lock. - **Database ahead of the target version**: the same, but backward, calling `unapply()` in descending order down to the target version. - **Another instance holds the lock**: sleep `migration_wait_sec` seconds and check again, until the migrations are complete or `migration_max_wait_sec` is exceeded, which raises a `MigrationTimeoutError`. A missing entry in the migration map raises a `NotImplementedError`, and a failing migration step raises a [`MigrationStepError`](../../reference/providers.mongodb.migrations.MigrationStepError.qmd) naming the version and direction that failed; in both cases the lock is released on context exit so other instances are not blocked forever. Since a failed instance raises rather than starting the service, deployment tooling can treat migration failures like any other startup failure. ## Checking the Database Version Sometimes a component needs to *verify* the database version without being allowed to change anything — a job that runs alongside the service, a readiness check, or a tool operating on the same database. The standalone [`check_db_version`](../../reference/providers.mongodb.migrations.check_db_version.qmd) helper reads the version collection and raises a [`DbVersionMismatchError`](../../reference/providers.mongodb.migrations.DbVersionMismatchError.qmd) if the current version differs from the expected one: ```python from hexkit.providers.mongodb.migrations import check_db_version await check_db_version(config=config, target_version=3) ``` ## Helpers for the hexkit v6 Upgrade With hexkit v6, the MongoDB provider switched from storing UUIDs and datetimes as strings to MongoDB's native BSON types (see [How DTOs Map to Documents](mongodb.qmd#how-dtos-map-to-documents)). Services upgrading across that boundary need a migration converting their stored documents, which is the same for everyone — so `hexkit.providers.mongodb.migrations.helpers` provides prefabricated change functions: - [`convert_uuids_and_datetimes_v6`](../../reference/providers.mongodb.migrations.helpers.convert_uuids_and_datetimes_v6.qmd) is a factory producing a change function that converts the named fields from strings to native UUIDs and from isoformat strings to UTC datetimes with millisecond precision. - [`convert_outbox_correlation_id_v6`](../../reference/providers.mongodb.migrations.helpers.convert_outbox_correlation_id_v6.qmd) converts the `__metadata__.correlation_id` of documents in [outbox collections](mongokafka.qmd) to a native UUID. - [`convert_persistent_event_v6`](../../reference/providers.mongodb.migrations.helpers.convert_persistent_event_v6.qmd) converts documents of a `PersistentKafkaPublisher` collection, including populating the `event_id` field introduced in v6. They plug directly into `migrate_docs_in_collection()`: ```python from hexkit.providers.mongodb.migrations import MigrationDefinition from hexkit.providers.mongodb.migrations.helpers import ( convert_uuids_and_datetimes_v6, ) from my_service.models import Upload class V2Migration(MigrationDefinition): """Migrate stored documents to the native types used by hexkit v6.""" version = 2 async def apply(self): convert = convert_uuids_and_datetimes_v6( uuid_fields=["upload_id"], date_fields=["created"] ) async with self.auto_finalize(coll_names=["uploads"], copy_indexes=True): await self.migrate_docs_in_collection( coll_name="uploads", change_function=convert, validation_model=Upload, id_field="upload_id", ) ``` For fields beyond the covered ones, wrap the helper in your own change function and apply additional modifications before returning the document. ## Testing Migrations Migrations are regular code and deserve tests. The [MongoDB test fixtures](mongodb.qmd#testing) give each test a disposable database: populate collections with documents in the old format, run the migration via a `MigrationManager`, and assert on the resulting documents. The framework's own [integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/migrations/test_mongodb_migrations.py) demonstrate this pattern extensively and double as runnable usage examples. ### MongoDB + Kafka ## Outbox Publisher A common challenge in microservice development is the need to replicate data between services without introducing tight coupling. The solution to this offered by hexkit is a special form of `MongoDbDao` grafted with a `KafkaEventPublisher`, called a `MongoKafkaDaoPublisher` ("Outbox Publisher"). Contrasted with the other provider in the `MongoKafka` subpackage, the Outbox Publisher is used in place of the basic `MongoDbDao` provider and event publishing occurs "in the background". In addition, instances are created via the factory class, `MongoKafkaDaoPublisherFactory`. More on that below. Like the Persistent Publisher, the Outbox Publisher also provides methods to publish stored events that haven't been published yet or to publish all events regardless of publish status. Using this provider requires you to install hexkit with both the `akafka` and `mongodb` extras. The Outbox Publisher is the implementation of the protocol-level [`DaoPublisher`](../protocols/daopublisher.qmd) interface; that chapter describes the backend-independent contract, while this section covers the MongoDB + Kafka specifics. ### Instantiation/Construction Construction via factory: Use `MongoKafkaDaoPublisherFactory.construct(...).get_dao(...)` with: - `name`: MongoDB collection name. - `dto_model`: The class representing the domain object, which should inherit from Pydantic's `BaseModel`. - `id_field`: Name of the ID field on the DTO (mapped to `_id` in MongoDB). This is used as the event key. - `dto_to_event(dto) -> JsonObject | None`: Maps the DTO to the event payload. Return `None` to skip publishing for a given change. - `event_topic`: Kafka topic where outbox events are emitted. - `autopublish` (default: True): If False, changes are stored first and can be published later using `publish_pending()`. - `fields_to_index` *not currently implemented.* ### Behavior Domain object state changes are captured by the Outbox Publisher and categorized as either an *upsertion* or a *deletion*. This binary categorization is reflected in the Kafka event types that the Outbox Publisher uses, which are hardcoded by hexkit: `CHANGE_EVENT_TYPE` ("upserted") and `DELETE_EVENT_TYPE` ("deleted"). But the Outbox Publisher is designed to be able to replay changes, so how are deletions preserved? This is accomplished through additional data that is automatically managed by hexkit and stored along with the domain data in the database. This data is stored in a field named `__metadata__` and tracks whether the data has been published as an event, what the event ID was for the last time the data was published, the correlation ID for the data, and a boolean indicating whether or not the data was deleted. When data managed by an Outbox Publisher is deleted, it is not entirely removed from the database. Instead, the ID field and metadata are kept (all other fields *are* deleted) and the `__metadata__.deleted` field is set to True. Even though the data still partially exists in the database, the Outbox Publisher will raise errors just as if the data did not exist. `get_by_id()` will raise a `ResourceNotFoundError` in this case. Metadata stored on documents: - `__metadata__.deleted` (bool): Whether the resource has been deleted. - `__metadata__.published` (bool): Whether the most recent change has been published to Kafka. - `__metadata__.correlation_id` (UUID): Correlation ID captured from context when the change was made; reused on (re)publish for tracing. - `__metadata__.last_event_id` (UUID | None): Event ID of the last emitted outbox event. When data is either added or modified deleted, the Outbox Publisher feeds the data into the Factory's `dto_to_event()` method to obtain the desired representation for an event. This might match the original data, or it might remove fields, e.g. to avoid transmitting superfluous information. If the result of `dto_to_event()` is `None`, no event is published. When data is deleted, the Outbox Publisher publishes an event with an empty payload but the key set to the ID. Consumers can subscribe to the given outbox topic and perform the suitable action based on whether the event type is "upserted" or "deleted" — hexkit's interface for this is the [DAO Subscriber](../protocols/daosubscriber.qmd) protocol. All event publishing depends on `autopublish` being set to True. Republishing events: - `publish_pending()`: Publishes all documents where `__metadata__.published == False` using the stored correlation ID and appropriate event type. - `republish()`: Republishes the current state of all documents (change events for non-deleted, tombstones for deleted), again using each document’s stored correlation ID. Correlation & Event IDs: - The current correlation ID is captured and stored on each write. - During (re)publish, the stored correlation ID is re-applied to the context so downstream consumers see a consistent trace. - Event IDs are generated every time the data is (re)published, and the ID is stored after the event is published. This is different from the Persistent Publisher in that the latter will reuse the same event ID. ### Usage example The following shows a contrived example of how to use the Outbox Publisher: ```python from pydantic import BaseModel from uuid import UUID from hexkit.providers.mongokafka import MongoKafkaDaoPublisherFactory, MongoKafkaConfig # Define a model class User(BaseModel): user_id: UUID name: str email: str # Determine how to represent the data as a Kafka event def user_to_event(user: User) -> dict | None: return { "user_id": str(user.user_id), "name": user.name, "email": user.email, } # Use MongoKafkaConfig to supply the Factory with the necessary information config = MongoKafkaConfig() # Construct the Factory as an async context manager async with MongoKafkaDaoPublisherFactory.construct(config=config) as factory: user_dao = await factory.get_dao( name="users", dto_model=User, id_field="user_id", dto_to_event=user_to_event, event_topic="users.outbox", autopublish=True, ) # Create or update data -> publishes a CHANGE event await user_dao.upsert( User( user_id=UUID("2e3975db-2c80-49f1-9f6b-cbb0174ca8f3"), name="Jane", email="jane@example.com", ) ) # Delete data -> publishes a DELETE tombstone await user_dao.delete(UUID("2e3975db-2c80-49f1-9f6b-cbb0174ca8f3")) # Operational controls await user_dao.publish_pending() # Drain any unpublished changes await user_dao.republish() # Replay full current state ``` ## Persistent Publisher With the basic Kafka publish provider (`KafkaEventPublisher`), event publication is not repeatable. The outbox publisher (`MongoKafkaDaoPublisher`) allows for a version of event publication, but mandates the use of a DAO in domain logic. There are plausible scenarios where the desired behavior is to publish events and persist them in a database via some transparent mechanism. For this, hexkit offers the `PersistentKafkaPublisher` ("Persistent Publisher"). The Persistent Publisher replaces existing usage of the `KafkaEventPublisher` and uses the MongoDB features exposed by hexkit to store a copy of each event as it is published, along with supplementary information such as when the event was published, whether the event was successfully published, the correlation ID associated with the event, and more. Like the Outbox Publisher, the Persistent Publisher also provides methods to publish stored events that haven't been published yet or to publish all events regardless of publish status. Using this provider requires you to install hexkit with both the `akafka` and `mongodb` extras. To learn more about the distinction between the Persistent Publisher and Outbox Publisher, see [this writeup](../developer_help/outbox_v_persistent_pub.qmd). ### Persistence Model The Persistent Publisher stores data according to the Pydantic model defined [here](../../reference/providers.mongokafka.PersistentKafkaEvent.qmd): ```python class PersistentKafkaEvent(BaseModel): """A model representing a kafka event to be published and stored in the database.""" compaction_key: str = Field( ..., description="The unique ID of the event. If the topic is set to be compacted," + " the ID is set to the topic and key in the format :. Otherwise" + " the ID is set to the actual event ID.", ) topic: Ascii = Field(..., description="The event topic") payload: JsonObject = Field(..., description="The event payload") key: Ascii = Field(..., description="The event key") type_: Ascii = Field(..., description="The event type") event_id: UUID4 | None = Field(default=None, description="The event ID") headers: Mapping[str, str] = Field( default_factory=dict, description="Non-standard event headers. Correlation ID, event_id, and event" + " type are transmitted as event headers, but added as such within the publisher" + " protocol. The headers here are any additional header that need to be sent.", ) correlation_id: UUID4 = UUID4Field(description="The event correlation ID") created: datetime = Field( ..., description="The timestamp of when the event record was first inserted" ) published: bool = Field(False, description="Whether the event has been published") ``` ### Instantiation/Construction The Persistent Publisher is meant to be used as an async context manager. The instance is created via the `construct()` method. There are three parameters unique to the Persistent Publisher: `compacted_topics`, `topics_not_stored`, and `collection_name`. - `compacted_topics` (set of strings): - Reflects remote topic compaction locally - Pass a set of topics as `compacted_topics` when constructing the publisher. - For those topics, the database uses a deterministic `compaction_key` of `:` as the document ID, so only the latest event per (topic, key) is stored. This mimics Kafka log compaction on the storage side and makes “republish latest” semantics trivial. - For non-compacted topics, the `compaction_key` is the `event_id` (a UUID), so every event is stored independently. This does mean that, for events in non-compacted topics, the event ID is stored as a UUID4 in the `event_id` field and as a string in the `compaction_key` field. - `topics_not_stored` (set of strings): - Provides a way to opt out of persistence for specified Kafka topics - Provide a set of topics that should be published but not stored. Those events are delegated to the plain `KafkaEventPublisher` that underpins the Persistent Publisher and will not appear in MongoDB. - `topics_not_stored` and `compacted_topics` must be disjoint; the provider will raise a `ValueError` if there is any overlap. - `collection_name`: - Determines the name of the collection used for storing events. All events are stored in the same collection. - By default, events are stored in the collection named `{service_name}PersistedEvents` (derived from `MongoKafkaConfig.service_name`). - The MongoDB document ID is `compaction_key`; this is unique by definition and makes writes idempotent for compacted topics. ### Stored information and Considerations: - Payload: - The event content, the payload, is stored in the field of the same name. - If developing a database migration that affects persisted events, take care to consider whether the payload itself needs to be updated as well. - Correlation IDs and headers: - If a correlation ID is present in the context, it’s captured and stored; otherwise, a new one is generated. - On (re)publish, the stored correlation ID is set in context so downstream consumers see a stable trace. - Custom headers you pass are persisted and sent on publish; standard headers (correlation ID, event ID, type) are handled by the publisher protocol. - Event ID and ordering: - The `created` field contains the timestamp denoting when the event was first published. It is not updated upon republishing. - `publish_pending` loads all documents with `published == false`, sorts by `created` ascending, and publishes them. This ensures that events are published in their original order. - `republish` iterates over all stored events; if an event lacks an `event_id`, it assigns one and marks `published = false` so the update is persisted. Then all events are published as above, regardless of whether they have been published already. - Events published to topics listed in `topics_not_stored` are never written to MongoDB, thus they won’t be affected by `publish_pending` or `republish`. - Considerations: - DLQ: Avoid using a Persistent Publisher as the DLQ Publisher for an event subscriber instance. Events should only ever be published once to the DLQ. - Idempotence: Think about the consumers of events published by the Persistent Publisher and utilize `compacted_topics` and `topics_not_stored` strategically. - Republishing is usually performed as a one-off command for a service, rather than somewhere in standard operation. But this is a convention, and the methods can be utilized as best fits a given use-case. ### Usage example in service 'abc' ```python from hexkit.providers.mongokafka import PersistentKafkaPublisher from hexkit.providers.mongodb import MongoDbDaoFactory from hexkit.providers.mongokafka import MongoKafkaConfig # Normally, topics would be defined in configuration. This is only for conciseness: COMPACTED_TOPICS = {"users"} TOPICS_NOT_STORED = {"notifications"} @asynccontextmanager async def get_persistent_publisher( config: MongoKafkaConfig, dao_factory: MongoDbDaoFactory | None = None ) -> AsyncGenerator[PersistentKafkaPublisher]: """Construct and return a PersistentKafkaPublisher.""" async with ( MongoDbDaoFactory.construct(config=config) as _dao_factory, PersistentKafkaPublisher.construct( config=config, dao_factory=_dao_factory, compacted_topics=COMPACTED_TOPICS, topics_not_stored=TOPICS_NOT_STORED collection_name="abcPersistedEvents", ) as persistent_publisher, ): yield persistent_publisher ``` ### S3 ## S3 Object Storage Provider The S3 provider implements the [Object Storage protocol](../protocols/objstorage.qmd) on top of the S3 API through [`S3ObjectStorage`](../../reference/providers.s3.S3ObjectStorage.qmd). It works with any S3-compatible backend: AWS S3 itself, [MinIO](https://min.io), [Ceph RGW](https://docs.ceph.com/en/latest/radosgw/), or [LocalStack](https://localstack.cloud) for local development and testing. Under the hood it uses the synchronous [boto3](https://boto3.amazonaws.com) client, with every blocking call wrapped in `asyncio.to_thread`, so all methods are safely awaitable without blocking the event loop. This section covers everything S3-specific; the backend-independent semantics of the operations (presigned URLs, multipart uploads, error types) are documented in the [Object Storage protocol](../protocols/objstorage.qmd) chapter. To use it, install hexkit with the `s3` extra: ```sh pip install hexkit[s3] ``` ### Configuration Connection parameters are collected in [`S3Config`](../../reference/providers.s3.S3Config.qmd), a `pydantic_settings.BaseSettings` class. Inherit your service's config class from it, or instantiate it directly: ```python from hexkit.providers.s3 import S3Config config = S3Config( s3_endpoint_url="http://localhost:4566", s3_access_key_id="my-access-key-id", s3_secret_access_key="my-secret-access-key", ) ``` | Parameter | Description | | --------- | ----------- | | `s3_endpoint_url` | The URL of the S3 API endpoint: an AWS regional endpoint, or the address of your MinIO, Ceph, or LocalStack instance. | | `s3_access_key_id` | The access key ID, as [used by AWS](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html). | | `s3_secret_access_key` | The secret access key belonging to the access key ID. Stored as a Pydantic `SecretStr` so it does not leak into logs or error messages. | | `s3_session_token` | Optional session token for temporary credentials, also stored as a `SecretStr`. Defaults to `None`. | | `aws_config_ini` | Optional path to an [AWS config INI file](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-a-configuration-file) for advanced botocore settings such as retry behavior or signature versions. Defaults to `None`. | ### Creating the Storage Client Unlike most other hexkit providers, `S3ObjectStorage` is created with a plain constructor rather than an async context manager: it does not hold open connections, so there is nothing to enter or close: ```python from hexkit.providers.s3 import S3ObjectStorage storage = S3ObjectStorage(config=config) await storage.create_bucket("my-bucket") ``` The resulting object implements the full [`ObjectStorageProtocol`](../../reference/protocols.objstorage.ObjectStorageProtocol.qmd) and can be passed anywhere the protocol is expected. ### Uploads, Downloads, and Presigned URLs All operations behave as described in the [Object Storage protocol](../protocols/objstorage.qmd) chapter. A condensed end-to-end session: ```python import httpx await storage.create_bucket("inbox") # Presigned POST URL for a single-request upload: presigned_post = await storage.get_object_upload_url( bucket_id="inbox", object_id="report-2026-07" ) # The client (not the service!) performs the actual upload: with open("report.pdf", "rb") as file: httpx.post( presigned_post.url, data=presigned_post.fields, files={"file": file}, ).raise_for_status() # Verify the outcome via metadata: assert await storage.does_object_exist( bucket_id="inbox", object_id="report-2026-07" ) size = await storage.get_object_size(bucket_id="inbox", object_id="report-2026-07") # Presigned URL for downloading (a plain GET URL): download_url = await storage.get_object_download_url( bucket_id="inbox", object_id="report-2026-07" ) await storage.delete_object(bucket_id="inbox", object_id="report-2026-07") ``` For the full operation catalog, including bucket management, object copying, and the multipart upload flow, see the [protocol chapter](../protocols/objstorage.qmd#uploading-objects). ### Multipart Uploads and Part Sizing The multipart upload flow follows the [protocol semantics](../protocols/objstorage.qmd#multipart-uploads). What the S3 API adds are hard limits: parts must be between 8 MiB and 4 GiB (except the last one), at most 10,000 parts are allowed per upload, and objects may not exceed 5 TiB. The [`calc_part_size`](../../reference/utils.calc_part_size.qmd) helper from `hexkit.utils` recommends a part size that respects all of these for a given file size: ```python from hexkit.utils import calc_part_size part_size = calc_part_size(file_size=file_size) # raises ValueError above 5 TiB upload_id = await storage.init_multipart_upload( bucket_id="inbox", object_id="large-dataset-001" ) # ...hand out part URLs and complete as described in the protocol chapter, # passing part_size as the anticipated_part_size. ``` You can also pass a `preferred_part_size`; it is returned unchanged if it satisfies the constraints for the given file size and adjusted otherwise. Note that `copy_object` uses `calc_part_size` internally, so server-side copies of large objects pick a suitable part size automatically. ### Ceph Tenant Buckets The provider relaxes the protocol's bucket naming rules to additionally accept Ceph RGW's multi-tenancy notation, where bucket names are prefixed with a tenant name and a colon, as in `my-tenant:my-bucket`. The part after the colon must still follow the usual bucket naming rules. ### Error Handling The provider translates botocore's `ClientError`s into the exception types of the [Object Storage protocol](../protocols/objstorage.qmd#error-handling), so calling code never needs to import from boto3 or botocore: | Situation | Raised exception | | --------- | ---------------- | | Accessing a bucket that does not exist | `BucketNotFoundError` | | `create_bucket` with an already existing bucket ID | `BucketAlreadyExistsError` | | `delete_bucket` on a non-empty bucket without `delete_content=True` | `BucketNotEmptyError` | | Accessing an object that does not exist | `ObjectNotFoundError` | | Upload URL or copy destination where the object already exists | `ObjectAlreadyExistsError` | | Multipart operation with an unknown upload ID | `MultiPartUploadNotFoundError` | | `init_multipart_upload` while another upload is active for the object | `MultiPartUploadAlreadyExistsError` | | `complete_multipart_upload` with missing or ill-sized parts | `MultiPartUploadConfirmError` | | `abort_multipart_upload` unable to remove all parts | `MultiPartUploadAbortError` | | Invalid bucket or object ID (raised before contacting S3) | `BucketIdValidationError` / `ObjectIdValidationError` | | Any other bucket-related S3 error | `BucketError` | | Any other object-related S3 error | `ObjectError` | | Any other S3 error | `ObjectStorageProtocolError` | Remember that these exception classes are nested on the `ObjectStorageProtocol` class, so you catch, for example, `ObjectStorageProtocol.ObjectNotFoundError`. One caveat: passing the `object_md5sum` argument to `does_object_exist` raises a `NotImplementedError`, as MD5 content checking is not implemented by this provider yet. The provider's own [integration tests](https://github.com/ghga-de/hexkit/blob/main/tests/integration/providers/objstorage/test_s3_objstorage.py) double as an extensive collection of runnable usage examples; the [S3 Object Storage Test Utils](#s3-object-storage-test-utils) section below describes the fixtures they build on. ## S3 KV Store Providers S3 providers implement `KeyValueStoreProtocol` using S3 objects, which is useful for large or opaque payloads. Available variants are `S3BytesKeyValueStore`, `S3StrKeyValueStore`, `S3JsonKeyValueStore`, and `S3DtoKeyValueStore`. API reference: [hexkit.providers.s3](../../reference/index.qmd#provider-s3) ## S3 Object Storage Test Utils For integration tests, the `test-s3` extra provides pytest fixtures based on [testcontainers](https://testcontainers-python.readthedocs.io) that spin up a disposable [LocalStack](https://localstack.cloud) container (image `localstack/localstack:4.0.3`) exposing a real S3 API, so tests exercise the actual provider without any external infrastructure: ```sh pip install hexkit[test-s3] ``` All utilities live in `hexkit.providers.s3.testutils`. ### Fixtures The `s3_fixture` yields an `S3Fixture` object (under the fixture name `s3`) that bundles a ready-to-use storage client with cleanup helpers. Both the container fixture and the storage fixture must be imported into the test module for pytest to resolve them, hence the `# noqa: F401` comments: ```python from hexkit.providers.s3.testutils import ( S3Fixture, s3_container_fixture, # noqa: F401 s3_fixture, # noqa: F401 ) async def test_upload(s3: S3Fixture): await s3.storage.create_bucket("test-bucket") ... ``` The fixtures split the work between two scopes: - `s3_container_fixture` (fixture name `s3_container`) is session-scoped: one LocalStack container is started per test session and reused by all tests. - `s3_fixture` (fixture name `s3`) is function-scoped and async: it deletes all buckets before each test, so every test starts from a clean storage. `clean_s3_fixture` is an alias for it. - `persistent_s3_fixture` is the variant without cleanup, for tests that intentionally share storage state. If the default scopes or names do not fit your test setup, the factory functions `get_s3_container_fixture`, `get_clean_s3_fixture`, and `get_persistent_s3_fixture` create the same fixtures with custom `scope` and `name` arguments. ### The S3Fixture An `S3Fixture` carries two attributes: `config`, an `S3Config` pointing at the test container, and `storage`, a fully configured `S3ObjectStorage`. On top of that it offers convenience methods for arranging test state: - `get_buckets()` lists all currently existing buckets. - `populate_buckets(buckets)` creates the given buckets. - `populate_file_objects(file_objects)` creates buckets as needed and uploads the given file objects. - `empty_buckets(buckets=..., exclude_buckets=...)` deletes all objects from the given buckets (or all buckets). - `delete_buckets(buckets=..., exclude_buckets=...)` deletes the given buckets (or all buckets) including their content. ### File Objects and Upload Helpers Test files are described by `FileObject`, a Pydantic model with `file_path`, `bucket_id`, and `object_id` fields plus computed `content` (the file's bytes) and `md5` properties. The `temp_file_object` context manager creates one backed by a temporary file of a given size, and the `tmp_file` fixture provides a default 5 MiB instance: ```python from hexkit.providers.s3.testutils import temp_file_object async def test_with_file(s3: S3Fixture): with temp_file_object("test-bucket", "test-object", size=1024) as file: await s3.populate_file_objects([file]) assert await s3.storage.does_object_exist( bucket_id="test-bucket", object_id="test-object" ) ``` For exercising transfers the way real clients perform them, several helpers issue actual HTTP requests against presigned URLs: `upload_file` POSTs a file to a `PresignedPostURL`, `multipart_upload_file` runs a complete multipart upload for a local file, `upload_part` sends a single part, and `calc_md5` computes checksums for integrity assertions. ### Compliance Testing with typical_workflow The `typical_workflow` coroutine runs an end-to-end scenario — create buckets, upload (single-request or multipart), verify, download and check integrity, copy to a second bucket, delete — against any `ObjectStorageProtocol` implementation. It serves both as a smoke test for your storage setup and as a compliance check when [implementing a custom provider](../protocols/objstorage.qmd#implementing-a-custom-provider). ### Multi-Storage Setups For services that span multiple independent object storages, the `s3_multi_container_fixture` starts several LocalStack containers keyed by alias (driven by a `storage_aliases` fixture you provide), and the `federated_s3_fixture` yields a `FederatedS3Fixture` holding one `S3Fixture` per alias. Persistent variants and factory functions exist analogously to the single-storage fixtures. API reference: [hexkit.providers.s3](../../reference/index.qmd#provider-s3) ### Redis Redis providers implement `KeyValueStoreProtocol` for low-latency key-value access. Available variants are `RedisBytesKeyValueStore`, `RedisStrKeyValueStore`, `RedisJsonKeyValueStore`, and `RedisDtoKeyValueStore`. API reference: [hexkit.providers.redis](../../reference/index.qmd#provider-redis) ### Vault Vault providers implement `KeyValueStoreProtocol` on top of Vault KV v2 and are suited for secrets and other sensitive values. Available variants are `VaultBytesKeyValueStore`, `VaultStrKeyValueStore`, `VaultJsonKeyValueStore`, and `VaultDtoKeyValueStore`. API reference: [hexkit.providers.vault](../../reference/index.qmd#provider-hashicorp-vault) ### In-Memory (Testing) In-memory providers implement `KeyValueStoreProtocol` for lightweight test setups without external infrastructure dependencies. Available variants are `InMemBytesKeyValueStore`, `InMemStrKeyValueStore`, `InMemJsonKeyValueStore`, and `InMemDtoKeyValueStore`. API reference: [hexkit.providers.testing](../../reference/index.qmd#provider-in-memory-testing) ## Observability Tools ### Observability Tools (intro here) ## Features - [Correlation IDs](correlation_ids.qmd) - [Configurable Logging](logging.qmd) ### Correlation IDs TBD ### Configurable Logging TBD ## Developer Help ### Developer Help (intro here) ## Guides - [Outbox vs Persistent Publisher: Choosing the Right One](outbox_v_persistent_pub.qmd) - [Getting the Most Out of the Test Fixtures](testing.qmd) ### Outbox vs Persistent Publisher: Choosing the Right One When using `KafkaEventPublisher`, produced events are generated and sent to the Kafka broker in an ephemeral manner: there's no ability to generate the same events again. The Persistent Publisher is a hexkit provider that solves this problem by storing a representation of events in MongoDB as they are published. This makes the database the de facto backup of Kafka, to an extent. The Persistent Publisher class is called `PersistentKafkaPublisher`, located in the `MongoKafka` subpackage. The other `MongoKafka` provider, the Outbox Publisher, solves a different problem. Microservices sometimes need to share data without sharing access to the same database. The Outbox Publisher is essentially a MongoDB DAO retrofitted with a Kafka publisher. It attaches a small amount of metadata to documents as it upserts or deletes data. Upon each modifying operation, the provider runs the given document through a user-defined function to produce a Kafka payload which it then publishes. If a document is deleted, the provider merely publishes a tombstone record with the object ID as the key. When another service wants to access the data in question, it subscribes to the Kafka topic housing the outbox events. The Outbox Publisher is called `MongoKafkaDaoPublisher`, also located in the `MongoKafka` subpackage. Both the `PersistentKafkaPublisher` and the `MongoKafkaDaoPublisher` save data to the database, but they are intended for use in different situations, and are not interchangeable. ### When to use the Persistent Publisher (`PersistentKafkaPublisher`) The Persistent Publisher should be employed when: - The publisher needs to be used like a normal `KafkaEventPublisher`. - The data published *does not* represent a stateful entity, like a domain object. - You want to be able to publish the same Kafka event again. - Same topic, event type, correlation ID, event ID, payload, etc. - Often these are events that initiate a sequence of events, or Kafka flow. ### When to use the Outbox Publisher (`MongoKafkaDaoPublisher`) The Outbox Publisher should be used when: - The publisher needs to be used like a normal `MongoDbDao`. - The data published represents the latest state of a domain object - Examples: user data, access requests, order status, etc. - Your service needs to inform other services about changes to a given MongoDB collection. - You're focused on **storing** data. The Outbox Publisher cannot publish arbitrary events like the Persistent Publisher — it behaves like the MongoDB DAO and publishes events in the background. ## Further Details ### How the Two Publishers Store Information The Persistent Publisher stores data according to the Pydantic model defined [here](../../reference/providers.mongokafka.PersistentKafkaEvent.qmd): ```python class PersistentKafkaEvent(BaseModel): """A model representing a kafka event to be published and stored in the database.""" compaction_key: str = Field( ..., description="The unique ID of the event. If the topic is set to be compacted," + " the ID is set to the topic and key in the format :. Otherwise" + " the ID is set to the actual event ID.", ) topic: Ascii = Field(..., description="The event topic") payload: JsonObject = Field(..., description="The event payload") key: Ascii = Field(..., description="The event key") type_: Ascii = Field(..., description="The event type") event_id: UUID4 | None = Field(default=None, description="The event ID") headers: Mapping[str, str] = Field( default_factory=dict, description="Non-standard event headers. Correlation ID, event_id, and event" + " type are transmitted as event headers, but added as such within the publisher" + " protocol. The headers here are any additional header that need to be sent.", ) correlation_id: UUID4 = UUID4Field(description="The event correlation ID") created: datetime = Field( ..., description="The timestamp of when the event record was first inserted" ) published: bool = Field(False, description="Whether the event has been published") ``` By contrast, the Outbox Publisher saves the domain object like a DAO normally would, but adds a `__metadata__` field. If we were saving data from a `User` class with a `user_id`, `name` and `email` field, the saved data might look like this: ```json { "__metadata__": { "correlation_id": , "published": true, "deleted": false, "last_event_id": }, "_id": , // hexkit converts models' ID field names to the MongoDB `_id` "name": "John Doe", "email": "doe@example.com" } ``` ### Republishing Both providers expose the methods `publish_pending` and `republish`. The former only publishes documents where `published` is false, while the latter will republish all its documents regardless of publish status. We suggest implementing a method in your service accessible via CLI in order to trigger republishing as a job. In both cases, the provider will use the stored correlation ID when republishing data. Please see the code below for an example of how to structure the republish method. ```python # In a module housing CLI commands: import asyncio from typing import Annotated from my_service.main import publish_events import typer cli = typer.Typer() @cli.command(name="publish-events") def sync_run_publish_events( all: Annotated[ bool, typer.Option(help="Set to (re)publish all events regardless of status") ] = False, ): """Publish pending events.""" asyncio.run(publish_events(all=all)) ``` ```python # In a module housing asynchronous entrypoint functions: from my_service.config import Config # Import the function that returns a configured PersistentKafkaPublisher or MongoKakfaDaoPublisher instance. # See the example code in the [Persistent Publisher](./persistent_publisher.qmd) documentation. from my_service.inject import get_persistent_publisher async def publish_events(*, all: bool = False): """Publish pending events. Set `--all` to (re)publish all events regardless of status.""" config = Config() async with get_persistent_publisher(config=config) as publisher: if all: await publisher.republish() else: await publisher.publish_pending() ``` ### Getting the Most Out of the Test Fixtures TBD ## Glossary ### Index # Glossary TBD