Quickstart¶

Even if you generate projects with apogee new, you will want to extend or override the base classes by hand. Everything below is exported from the package root, so from apogee_core import X is all you need.

Define an entity¶

BaseEntity is a plain dataclass base — no ORM, no metaclass — giving every entity an id, created_at and updated_at, plus to_dict().

domain/entities/customer.py

from dataclasses import dataclass

from apogee_core import BaseEntity


@dataclass
class Customer(BaseEntity):
    name: str = ""
    email: str = ""

Fields need defaults because BaseEntity already declares defaulted fields — a dataclass cannot place a required field after a defaulted one.

Raise domain errors, not HTTP errors¶

handle_domain_exception translates these into responses, so the domain layer never imports a web framework.

domain/exceptions/customer_exceptions.py

from apogee_core import AlreadyExistsException, NotFoundException


class CustomerNotFound(NotFoundException):
    """Raised when a customer id has no match."""


class CustomerAlreadyExists(AlreadyExistsException):
    """Raised when the email is already registered."""

DomainException is the shared base, if you need to catch all of them.

Wrap the response¶

ApiResponse is a factory with three entry points.

presentation/http/controllers/customer_controller.py

from apogee_core import ApiResponse

ApiResponse.success(customer)                     # -> SuccessResponse[Customer]
ApiResponse.paged(page)                           # -> PagedResponse
ApiResponse.error(code="not_found", message="No such customer")   # -> ErrorResponse

Query with filters, search and paging¶

QueryFilter is the engine-agnostic query contract — a frozen Pydantic model that each repository translates to SQL or to a MongoDB query.

application/services/customer_service.py

from apogee_core import PagedResult, QueryFilter

query = QueryFilter(
    page=1,
    per_page=20,
    filters={"active": True},
    search="ada",
    search_fields=("name", "email"),
    sort=(("created_at", "desc"),),
)

result: PagedResult[Customer] = await repository.find_all(query)
print(result.total, len(result.data))

The narrower IQueryRepository port exposes get_by_id(entity_id) and get_all(filters, page, per_page); IFilterableQueryRepository adds the QueryFilter path shown above.

Pick a repository base¶

Base class	Engine	Port it satisfies
`BaseSQLQueryRepository` / `BaseSQLCommandRepository`	PostgreSQL, MySQL (SQLAlchemy async)	`IQueryRepository`, `ICommandRepository`
`BaseMongoQueryRepository` / `BaseMongoCommandRepository`	MongoDB	`IMongoQueryRepository`, `IMongoCommandRepository`

The SQL base is generic over the entity and the ORM model, and it allow-lists the fields a client may filter, sort or search on — an unlisted field is rejected rather than passed to the database.

infrastructure/persistence/customer_repository.py

from apogee_core import BaseSQLQueryRepository


class CustomerQueryRepository(BaseSQLQueryRepository[Customer, CustomerModel]):
    _model_class = CustomerModel

    ALLOWED_FILTER_FIELDS = frozenset({"active", "country"})
    ALLOWED_SORT_FIELDS = frozenset({"created_at", "name"})
    SEARCHABLE_FIELDS = frozenset({"name", "email"})

    def _to_entity(self, model: CustomerModel) -> Customer:
        return Customer(id=model.id, name=model.name, email=model.email)

Add a cache¶

ICacheStore ships with two implementations, and SqlCacheAside is a mixin that layers read-through caching onto a SQL query repository.

Python

from apogee_core import InMemoryCacheStore, SqlCacheAside


class CachedCustomerRepository(SqlCacheAside, CustomerQueryRepository):
    cache_ttl_s = 60.0