Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt

Use this file to discover all available pages before exploring further.

Domain services in HexCore handle business operations that span multiple entities or require coordination that does not naturally belong to a single entity. BaseDomainService is the base class for all domain services. Beyond providing access to the configured EventBus, it ships with a powerful list_entities() method that handles filtering, sorting, pagination, and full-text search out of the box — driven by a declarative QueryRequestDTO. This page documents the class, all query DTOs, and shows a complete service example.

​
BaseDomainService

BaseDomainService lives in hexcore.domain.services. 
from hexcore.domain.services import BaseDomainService

​
Constructor

​
event_bus
EventBus
default:"LazyConfig().get_config().event_bus"
The active EventBus instance. Defaults to the bus configured in ServerConfig. Override this in tests or specialised setups by passing a custom EventBus directly.
After construction the service also stores self.config = LazyConfig.get_config() so subclasses have access to all project settings without an extra import.

​
list_entities()

async def list_entities(
    self,
    repository: IBaseRepository[T],
    query: QueryRequestDTO,
) -> QueryResponseDTO:
list_entities is the primary query engine. It inspects the repository for an optional query_all(query) method:
  • If query_all exists on the repository (e.g. a SQLAlchemy implementation that pushes filters to SQL), it is called directly and the results are wrapped in a QueryResponseDTO.
  • If query_all does not exist, list_entities calls repository.list_all(limit=None, offset=0) to fetch all entities in-memory, then applies search, filtering, sorting, and pagination via query_entities().
Implement query_all(query: QueryRequestDTO) -> tuple[list[T], int] on your repository to push filtering and sorting to the database and avoid loading all rows into memory.

​
Query DTOs

​
QueryRequestDTO

from hexcore.application.dtos.query import QueryRequestDTO
​
limit
int
default:"50"
Maximum number of items to return. Must be ≥ 1.
​
offset
int
default:"0"
Number of items to skip before collecting results. Must be ≥ 0.
​
search
str | None
default:"None"
Full-text search string. When set, only entities that contain this string (case-insensitive) in any of the search_fields are returned. If search_fields is empty, HexCore infers searchable fields automatically from scalar field types (str, int, float, bool, date, datetime).
​
search_fields
list[str]
default:"[]"
List of entity field names to search within. Leave empty to search all scalar fields.
​
filters
list[FilterConditionDTO]
default:"[]"
List of filter conditions applied with AND logic. All conditions must match for an entity to be included.
​
sort
list[SortConditionDTO]
default:"[]"
List of sort conditions applied in priority order. The first condition is the primary sort key (highest precedence); each subsequent condition is a tiebreaker. Multi-key sort is stable.

​
QueryResponseDTO

from hexcore.application.dtos.query import QueryResponseDTO
​
items
list[Any]
The page of entities matching the query.
​
total
int
Total number of entities that match the query (before pagination).
​
limit
int
The limit value echoed from the request.
​
offset
int
The offset value echoed from the request.
​
has_next
bool
True when there are more items beyond the current page.

​
FilterConditionDTO

from hexcore.application.dtos.query import FilterConditionDTO, FilterOperator
​
field
str
Name of the entity attribute to filter on. Must be non-empty.
​
operator
FilterOperator
default:"FilterOperator.EQ"
Comparison operator. See FilterOperator below.
​
value
Any
default:"None"
Value to compare the field against. For IS_NULL, this field is ignored.

​
FilterOperator

ValueStringDescription
EQ"eq"Equal to
NE"ne"Not equal to
GT"gt"Greater than
GTE"gte"Greater than or equal
LT"lt"Less than
LTE"lte"Less than or equal
IN"in"Value is in the given list
NOT_IN"not_in"Value is not in the given list
CONTAINS"contains"String field contains the given substring (case-insensitive)
STARTSWITH"startswith"String field starts with the given prefix (case-insensitive)
ENDSWITH"endswith"String field ends with the given suffix (case-insensitive)
IS_NULL"is_null"Field is None

​
SortConditionDTO

from hexcore.application.dtos.query import SortConditionDTO, SortDirection
​
field
str
Name of the entity attribute to sort by. Must be non-empty.
​
direction
SortDirection
default:"SortDirection.ASC"
Sort direction: SortDirection.ASC ("asc") or SortDirection.DESC ("desc").

​
Complete service example

The example below defines a ProductService with three real-world methods: list_products (paged query with filters), create_product, and deactivate_product. 
# src/domain/products/entities.py
from __future__ import annotations
from decimal import Decimal
from datetime import datetime, UTC
from hexcore.domain.base import BaseEntity
from hexcore.domain.events import EntityCreatedEvent, EntityDeletedEvent


class ProductCreatedEvent(EntityCreatedEvent["Product"]):
    pass


class ProductDeactivatedEvent(EntityDeletedEvent):
    pass


class Product(BaseEntity):
    name: str
    description: str = ""
    price: Decimal
    stock: int = 0
    category: str

    @classmethod
    def create(
        cls, name: str, price: Decimal, category: str, stock: int = 0
    ) -> "Product":
        product = cls(
            name=name, price=price, category=category, stock=stock
        )
        product.register_event(
            ProductCreatedEvent(entity_id=product.id, entity_data=product)
        )
        return product

    async def deactivate_product(self) -> None:
        await self.deactivate()
        self.register_event(
            ProductDeactivatedEvent(entity_id=self.id)
        )

​
Composing queries

Build QueryRequestDTO programmatically for complex query scenarios: 
from hexcore.application.dtos.query import (
    QueryRequestDTO,
    FilterConditionDTO,
    FilterOperator,
    SortConditionDTO,
    SortDirection,
)

query = QueryRequestDTO(
    # Full-text search across name and category
    search="wireless",
    search_fields=["name", "description", "category"],

    # Combine multiple filters (AND logic)
    filters=[
        FilterConditionDTO(field="is_active", operator=FilterOperator.EQ, value=True),
        FilterConditionDTO(field="stock", operator=FilterOperator.GT, value=0),
        FilterConditionDTO(
            field="category",
            operator=FilterOperator.IN,
            value=["electronics", "accessories"],
        ),
    ],

    # Sort by price ascending, then name alphabetically
    sort=[
        SortConditionDTO(field="price", direction=SortDirection.ASC),
        SortConditionDTO(field="name", direction=SortDirection.ASC),
    ],

    limit=25,
    offset=0,
)

response = await product_service.list_entities(repo, query)

print(f"Found {response.total} products, showing {len(response.items)}")
print(f"Has next page: {response.has_next}")

Build docs developers (and LLMs) love

Get started for freeTalk to us