Skip to content

API reference

Generated from the apogee-core source with mkdocstrings. Every symbol below is exported from apogee_core, so it is part of the supported public surface.

Other

ApiResponse

success staticmethod

Python
success(data: T) -> SuccessResponse[T]
Source code in apogee_core/presentation/http/responses/base_api_response.py
Python
@staticmethod
def success(data: T) -> SuccessResponse[T]:
    return SuccessResponse(data=data)

paged staticmethod

Python
paged(result) -> PagedResponse
Source code in apogee_core/presentation/http/responses/base_api_response.py
Python
@staticmethod
def paged(result) -> PagedResponse:
    return PagedResponse(
        data=result.data,
        meta=PageMeta(
            total=result.total,
            page=result.page,
            per_page=result.per_page,
            total_pages=result.total_pages,
        ),
    )

error staticmethod

Python
error(code: str, message: str) -> ErrorResponse
Source code in apogee_core/presentation/http/responses/base_api_response.py
Python
@staticmethod
def error(code: str, message: str) -> ErrorResponse:
    return ErrorResponse(error=ErrorDetail(code=code, message=message))

validation_error staticmethod

Python
validation_error(fields: list[dict]) -> ErrorResponse
Source code in apogee_core/presentation/http/responses/base_api_response.py
Python
@staticmethod
def validation_error(fields: list[dict]) -> ErrorResponse:
    return ErrorResponse(
        error=ErrorDetail(code="VALIDATION_ERROR", message="One or more fields are invalid.", fields=fields)
    )

BaseController

Bases: ABC

Base class for HTTP controllers. Provides common helpers.

success staticmethod

Python
success(data: Any) -> SuccessResponse
Source code in apogee_core/presentation/http/controllers/base_controller.py
Python
@staticmethod
def success(data: Any) -> SuccessResponse:
    return ApiResponse.success(data)

paged staticmethod

Python
paged(result: PagedResult) -> PagedResponse
Source code in apogee_core/presentation/http/controllers/base_controller.py
Python
@staticmethod
def paged(result: PagedResult) -> PagedResponse:
    return ApiResponse.paged(result)

parse_page_params staticmethod

Python
parse_page_params(params: dict) -> tuple[int, int]

Extract page and per_page from query params with defaults.

Source code in apogee_core/presentation/http/controllers/base_controller.py
Python
@staticmethod
def parse_page_params(params: dict) -> tuple[int, int]:
    """Extract page and per_page from query params with defaults."""
    page = max(1, int(params.get("page", "1")))
    per_page = min(100, max(1, int(params.get("per_page", "20"))))
    return page, per_page

BaseEntity dataclass

Python
BaseEntity(id: UUID = uuid4(), created_at: datetime = (lambda: now(utc))(), updated_at: datetime = (lambda: now(utc))())

id class-attribute instance-attribute

Python
id: UUID = field(default_factory=uuid4)

created_at class-attribute instance-attribute

Python
created_at: datetime = field(default_factory=lambda: now(utc))

updated_at class-attribute instance-attribute

Python
updated_at: datetime = field(default_factory=lambda: now(utc))

to_dict

Python
to_dict() -> dict[str, Any]
Source code in apogee_core/domain/entities/base_entity.py
Python
def to_dict(self) -> dict[str, Any]:
    result = {}
    for key, value in asdict(self).items():
        result[key] = self._serialize_value(value)
    return result

BaseMongoCommandRepository

Python
BaseMongoCommandRepository(database: Any)

Bases: Generic[T]

Subclasses must define:

  • _collection_name (str)
  • _to_document(entity: T) -> dict for serialisation
  • _to_entity(doc: dict) -> T for hydration
  • _id_of(entity: T) -> UUID to extract the primary key
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_command_repository.py
Python
def __init__(self, database: Any) -> None:
    self._db = database
    self._collection = database[self._collection_name]

create async

Python
create(entity: T) -> T
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_command_repository.py
Python
async def create(self, entity: T) -> T:
    doc = self._to_document(entity)
    doc.setdefault("_id", str(self._id_of(entity)))
    await self._collection.insert_one(doc)
    return entity

update async

Python
update(entity: T) -> T
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_command_repository.py
Python
async def update(self, entity: T) -> T:
    doc = self._to_document(entity)
    entity_id = str(self._id_of(entity))
    doc["_id"] = entity_id
    await self._collection.replace_one({"_id": entity_id}, doc, upsert=False)
    return entity

delete async

Python
delete(entity_id: UUID) -> bool
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_command_repository.py
Python
async def delete(self, entity_id: UUID) -> bool:
    result = await self._collection.delete_one({"_id": str(entity_id)})
    return bool(result.deleted_count)

upsert async

Python
upsert(entity: T) -> T
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_command_repository.py
Python
async def upsert(self, entity: T) -> T:
    doc = self._to_document(entity)
    entity_id = str(self._id_of(entity))
    doc["_id"] = entity_id
    await self._collection.replace_one({"_id": entity_id}, doc, upsert=True)
    return entity

BaseMongoQueryRepository

Python
BaseMongoQueryRepository(database: Any)

Bases: Generic[T]

Generic Mongo query repo. Subclasses must define:

  • _collection_name (str)
  • ALLOWED_FILTER_FIELDS, ALLOWED_SORT_FIELDS, SEARCHABLE_FIELDS
  • _to_entity(doc: dict) -> T for hydration
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_query_repository.py
Python
def __init__(self, database: Any) -> None:
    # ``database`` is a motor.motor_asyncio.AsyncIOMotorDatabase. Kept as
    # ``Any`` to avoid an import-time dependency on motor; we only touch
    # async methods that exist on motor's API.
    self._db = database
    self._collection = database[self._collection_name]

ALLOWED_FILTER_FIELDS class-attribute instance-attribute

Python
ALLOWED_FILTER_FIELDS: frozenset[str] = frozenset()

ALLOWED_SORT_FIELDS class-attribute instance-attribute

Python
ALLOWED_SORT_FIELDS: frozenset[str] = frozenset()

SEARCHABLE_FIELDS class-attribute instance-attribute

Python
SEARCHABLE_FIELDS: frozenset[str] = frozenset()

get_by_id async

Python
get_by_id(entity_id: UUID) -> T | None
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_query_repository.py
Python
async def get_by_id(self, entity_id: UUID) -> T | None:
    doc = await self._collection.find_one({"_id": str(entity_id)})
    return self._to_entity(doc) if doc else None

count async

Python
count(filters: dict[str, Any] | None = None) -> int
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_query_repository.py
Python
async def count(self, filters: dict[str, Any] | None = None) -> int:
    return int(await self._collection.count_documents(filters or {}))

find_all async

Python
find_all(query: QueryFilter | None = None) -> PagedResult[T]
Source code in apogee_core/infrastructure/persistence/mongo/base_mongo_query_repository.py
Python
async def find_all(self, query: QueryFilter | None = None) -> PagedResult[T]:
    q = query or QueryFilter()
    mongo_filter = query_filter_to_mongo(q, self.ALLOWED_FILTER_FIELDS | self.SEARCHABLE_FIELDS)
    total = int(await self._collection.count_documents(mongo_filter))
    skip = (q.page - 1) * q.per_page
    cursor = self._collection.find(mongo_filter)
    sort_spec = query_sort_to_mongo(q, self.ALLOWED_SORT_FIELDS)
    if sort_spec:
        cursor = cursor.sort(sort_spec)
    cursor = cursor.skip(skip).limit(q.per_page)
    if q.fields:
        projection = {field: 1 for field in q.fields if field in self.ALLOWED_FILTER_FIELDS}
        if projection:
            cursor = cursor.projection(projection)
    docs = await cursor.to_list(length=q.per_page)
    return PagedResult[T](
        data=[self._to_entity(d) for d in docs],
        total=total,
        page=q.page,
        per_page=q.per_page,
    )

BaseSQLCommandRepository

Bases: Generic[T, M]

BaseSQLCommandRepository — generic CRUD mixin for SQLAlchemy async.

create async

Python
create(entity: T) -> T
Source code in apogee_core/infrastructure/persistence/sql/base_sql_command_repository.py
Python
async def create(self, entity: T) -> T:
    async with self._session_factory() as session:
        model = self._to_model(entity)
        session.add(model)
        await session.commit()
        await session.refresh(model)
        return self._to_entity(model)

update async

Python
update(entity: T) -> T
Source code in apogee_core/infrastructure/persistence/sql/base_sql_command_repository.py
Python
async def update(self, entity: T) -> T:
    async with self._session_factory() as session:
        result = await session.execute(
            select(self._model_class).where(self._model_class.id == str(entity.id))
        )
        model = result.scalar_one_or_none()
        if not model:
            raise ValueError(f"{self._model_class.__name__} not found")
        for key, value in self._to_model(entity).__dict__.items():
            if key.startswith("_"):
                continue
            setattr(model, key, value)
        await session.commit()
        await session.refresh(model)
        return self._to_entity(model)

delete async

Python
delete(entity_id: UUID) -> bool
Source code in apogee_core/infrastructure/persistence/sql/base_sql_command_repository.py
Python
async def delete(self, entity_id: UUID) -> bool:
    async with self._session_factory() as session:
        result = await session.execute(
            select(self._model_class).where(self._model_class.id == str(entity_id))
        )
        model = result.scalar_one_or_none()
        if model:
            await session.delete(model)
            await session.commit()
            return True
        return False

BaseSQLQueryRepository

Bases: Generic[T, M]

BaseSQLQueryRepository — generic find_all for SQLAlchemy async.

ALLOWED_FILTER_FIELDS instance-attribute

Python
ALLOWED_FILTER_FIELDS: frozenset[str]

ALLOWED_SORT_FIELDS instance-attribute

Python
ALLOWED_SORT_FIELDS: frozenset[str]

SEARCHABLE_FIELDS instance-attribute

Python
SEARCHABLE_FIELDS: frozenset[str]

find_all async

Python
find_all(query: QueryFilter) -> PagedResult[T]
Source code in apogee_core/infrastructure/persistence/sql/base_sql_query_repository.py
Python
async def find_all(self, query: QueryFilter) -> PagedResult[T]:
    async with self._session_factory() as session:
        stmt = self._build_select(query)
        count_stmt = select(func.count()).select_from(stmt.subquery())
        total = (await session.execute(count_stmt)).scalar_one()
        result = await session.execute(stmt.offset(query.skip).limit(query.limit))
        models = result.scalars().all()
        return PagedResult(
            data=[self._to_entity(m) for m in models],
            total=total,
            page=query.page,
            per_page=query.per_page,
        )

find_by_id async

Python
find_by_id(entity_id: UUID) -> Optional[T]
Source code in apogee_core/infrastructure/persistence/sql/base_sql_query_repository.py
Python
async def find_by_id(self, entity_id: UUID) -> Optional[T]:
    async with self._session_factory() as session:
        result = await session.execute(
            select(self._model_class).where(self._model_class.id == str(entity_id))
        )
        model = result.scalar_one_or_none()
        return self._to_entity(model) if model else None

count async

Python
count() -> int
Source code in apogee_core/infrastructure/persistence/sql/base_sql_query_repository.py
Python
async def count(self) -> int:
    async with self._session_factory() as session:
        result = await session.execute(select(func.count()).select_from(self._model_class))
        return result.scalar_one()

ErrorDetail

Bases: BaseModel

code instance-attribute

Python
code: str

message instance-attribute

Python
message: str

fields class-attribute instance-attribute

Python
fields: Optional[list[dict]] = None

ErrorResponse

Bases: BaseModel

success class-attribute instance-attribute

Python
success: bool = False

error instance-attribute

Python
error: ErrorDetail

GenericCrudController

Bases: Generic[TRequestCreate, TRequestUpdate, TCreateDTO, TUpdateDTO], BaseController

Controlador genérico que abstrai operações CRUD padrão. Subclasses devem implementar _map_create_request_to_dto e _map_update_request_to_dto.

create async classmethod

Python
create(request: TRequestCreate, use_case: Any) -> SuccessResponse
Source code in apogee_core/presentation/http/controllers/generic_controller.py
Python
@classmethod
async def create(cls, request: TRequestCreate, use_case: Any) -> SuccessResponse:
    dto = cls._map_create_request_to_dto(request)
    return cls.success(await use_case.execute(dto))

get async classmethod

Python
get(entity_id: str, use_case: Any) -> SuccessResponse
Source code in apogee_core/presentation/http/controllers/generic_controller.py
Python
@classmethod
async def get(cls, entity_id: str, use_case: Any) -> SuccessResponse:
    return cls.success(await use_case.execute(entity_id))

list_all async classmethod

Python
list_all(use_case: Any, query: QueryFilter) -> PagedResponse
Source code in apogee_core/presentation/http/controllers/generic_controller.py
Python
@classmethod
async def list_all(cls, use_case: Any, query: QueryFilter) -> PagedResponse:
    return cls.paged(await use_case.execute(query))

update async classmethod

Python
update(entity_id: str, request: TRequestUpdate, use_case: Any) -> SuccessResponse
Source code in apogee_core/presentation/http/controllers/generic_controller.py
Python
@classmethod
async def update(cls, entity_id: str, request: TRequestUpdate, use_case: Any) -> SuccessResponse:
    dto = cls._map_update_request_to_dto(request)
    return cls.success(await use_case.execute(entity_id, dto))

delete async classmethod

Python
delete(entity_id: str, use_case: Any) -> SuccessResponse
Source code in apogee_core/presentation/http/controllers/generic_controller.py
Python
@classmethod
async def delete(cls, entity_id: str, use_case: Any) -> SuccessResponse:
    return cls.success(await use_case.execute(entity_id))

GenericCrudService

Python
GenericCrudService(command_repository: ICommandRepository[TEntity], query_repository: IQueryRepository[TEntity])

Bases: Generic[TEntity, TCreateDTO, TUpdateDTO, TOutputDTO]

Source code in apogee_core/application/services/generic_service.py
Python
def __init__(
    self,
    command_repository: ICommandRepository[TEntity],
    query_repository: IQueryRepository[TEntity],
) -> None:
    self.command_repository = command_repository
    self.query_repository = query_repository

command_repository instance-attribute

Python
command_repository = command_repository

query_repository instance-attribute

Python
query_repository = query_repository

create async

Python
create(dto: TCreateDTO) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def create(self, dto: TCreateDTO) -> TOutputDTO:
    entity = self._to_entity(dto)
    created = await self.command_repository.create(entity)
    return self._to_output(created)

get async

Python
get(entity_id: str) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def get(self, entity_id: str) -> TOutputDTO:
    entity = await self.query_repository.find_by_id(UUID(entity_id))
    if not entity:
        raise self._not_found_exception(entity_id)
    return self._to_output(entity)

list async

Python
list(query: QueryFilter) -> PagedResult[TOutputDTO]
Source code in apogee_core/application/services/generic_service.py
Python
async def list(self, query: QueryFilter) -> PagedResult[TOutputDTO]:
    result = await self.query_repository.find_all(query)
    return PagedResult(
        data=[self._to_output(e) for e in result.data],
        total=result.total,
        page=result.page,
        per_page=result.per_page,
    )

update async

Python
update(entity_id: str, dto: TUpdateDTO) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def update(self, entity_id: str, dto: TUpdateDTO) -> TOutputDTO:
    entity = await self.query_repository.find_by_id(UUID(entity_id))
    if not entity:
        raise self._not_found_exception(entity_id)

    self._update_entity_from_dto(entity, dto)
    entity.updated_at = datetime.now(timezone.utc)
    updated = await self.command_repository.update(entity)
    return self._to_output(updated)

delete async

Python
delete(entity_id: str) -> bool
Source code in apogee_core/application/services/generic_service.py
Python
async def delete(self, entity_id: str) -> bool:
    entity = await self.query_repository.find_by_id(UUID(entity_id))
    if not entity:
        raise self._not_found_exception(entity_id)
    return await self.command_repository.delete(entity.id)

HttpStatus

Bases: IntEnum

OK class-attribute instance-attribute

Python
OK = 200

CREATED class-attribute instance-attribute

Python
CREATED = 201

NO_CONTENT class-attribute instance-attribute

Python
NO_CONTENT = 204

BAD_REQUEST class-attribute instance-attribute

Python
BAD_REQUEST = 400

UNAUTHORIZED class-attribute instance-attribute

Python
UNAUTHORIZED = 401

FORBIDDEN class-attribute instance-attribute

Python
FORBIDDEN = 403

NOT_FOUND class-attribute instance-attribute

Python
NOT_FOUND = 404

CONFLICT class-attribute instance-attribute

Python
CONFLICT = 409

UNPROCESSABLE class-attribute instance-attribute

Python
UNPROCESSABLE = 422

INTERNAL_ERROR class-attribute instance-attribute

Python
INTERNAL_ERROR = 500

InMemoryCacheStore

Python
InMemoryCacheStore(namespace: str = '')
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
def __init__(self, namespace: str = "") -> None:
    self._namespace = namespace
    self._data: dict[str, _Entry] = {}

name class-attribute instance-attribute

Python
name = 'in_memory'

get async

Python
get(key: str) -> Any | None
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def get(self, key: str) -> Any | None:
    entry = self._data.get(self._key(key))
    if entry is None:
        return None
    if not self._is_alive(entry):
        self._data.pop(self._key(key), None)
        return None
    return entry.value

set async

Python
set(key: str, value: Any, ttl_s: float | None = None) -> None
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def set(self, key: str, value: Any, ttl_s: float | None = None) -> None:
    expires = (time.monotonic() + ttl_s) if ttl_s is not None else None
    self._data[self._key(key)] = _Entry(value=value, expires_at=expires)

delete async

Python
delete(key: str) -> bool
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def delete(self, key: str) -> bool:
    return self._data.pop(self._key(key), None) is not None

exists async

Python
exists(key: str) -> bool
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def exists(self, key: str) -> bool:
    entry = self._data.get(self._key(key))
    if entry is None:
        return False
    if not self._is_alive(entry):
        self._data.pop(self._key(key), None)
        return False
    return True

expire async

Python
expire(key: str, ttl_s: float) -> bool
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def expire(self, key: str, ttl_s: float) -> bool:
    entry = self._data.get(self._key(key))
    if entry is None or not self._is_alive(entry):
        return False
    entry.expires_at = time.monotonic() + ttl_s
    return True

clear async

Python
clear() -> None
Source code in apogee_core/infrastructure/cache/in_memory_cache_store.py
Python
async def clear(self) -> None:
    if not self._namespace:
        self._data.clear()
        return
    prefix = f"{self._namespace}:"
    for key in [k for k in self._data if k.startswith(prefix)]:
        del self._data[key]

PageMeta

Bases: BaseModel

total instance-attribute

Python
total: int

page instance-attribute

Python
page: int

per_page instance-attribute

Python
per_page: int

total_pages instance-attribute

Python
total_pages: int

PagedResponse

Bases: BaseModel, Generic[T]

success class-attribute instance-attribute

Python
success: bool = True

data instance-attribute

Python
data: list[T]

meta instance-attribute

Python
meta: PageMeta

PagedResult

Bases: BaseModel, Generic[T]

data instance-attribute

Python
data: List[T]

total instance-attribute

Python
total: int

page instance-attribute

Python
page: int

per_page instance-attribute

Python
per_page: int

model_config class-attribute instance-attribute

Python
model_config = {'frozen': True}

total_pages property

Python
total_pages: int

has_next property

Python
has_next: bool

has_prev property

Python
has_prev: bool

QueryFilter

Bases: BaseModel

page class-attribute instance-attribute

Python
page: int = 1

per_page class-attribute instance-attribute

Python
per_page: int = 20

search class-attribute instance-attribute

Python
search: Optional[str] = None

search_fields class-attribute instance-attribute

Python
search_fields: tuple[str, ...] = ()

sort class-attribute instance-attribute

Python
sort: tuple[tuple[str, str], ...] = ()

fields class-attribute instance-attribute

Python
fields: tuple[str, ...] = ()

filters class-attribute instance-attribute

Python
filters: dict[str, Any] = {}

skip property

Python
skip: int

limit property

Python
limit: int

page_must_be_positive classmethod

Python
page_must_be_positive(v: int) -> int
Source code in apogee_core/domain/value_objects/query_filter_vo.py
Python
@field_validator("page")
@classmethod
def page_must_be_positive(cls, v: int) -> int:
    if v < 1:
        raise ValueError("page must be >= 1")
    return v

per_page_in_range classmethod

Python
per_page_in_range(v: int) -> int
Source code in apogee_core/domain/value_objects/query_filter_vo.py
Python
@field_validator("per_page")
@classmethod
def per_page_in_range(cls, v: int) -> int:
    if v < 1 or v > 1000:
        raise ValueError("per_page must be between 1 and 1000")
    return v

from_params classmethod

Python
from_params(*, page: int = 1, per_page: int = 20, search: Optional[str] = None, search_fields: tuple[str, ...] = (), sort: Optional[str] = None, fields: Optional[str] = None, filters: Optional[dict[str, Any]] = None) -> 'QueryFilter'
Source code in apogee_core/domain/value_objects/query_filter_vo.py
Python
@classmethod
def from_params(
    cls,
    *,
    page: int = 1,
    per_page: int = 20,
    search: Optional[str] = None,
    search_fields: tuple[str, ...] = (),
    sort: Optional[str] = None,
    fields: Optional[str] = None,
    filters: Optional[dict[str, Any]] = None,
) -> "QueryFilter":
    return cls(
        page=max(1, page),
        per_page=max(1, min(per_page, 1000)),
        search=search or None,
        search_fields=search_fields,
        sort=_parse_sort(sort) if sort else (),
        fields=_parse_fields(fields) if fields else (),
        filters=filters or {},
    )

QueryFilterBuilder

build staticmethod

Python
build(*, page: int = 1, per_page: int = 20, q: Optional[str] = None, sort: Optional[str] = None, fields: Optional[str] = None, search_fields: frozenset[str] = frozenset(), raw_params: Optional[dict[str, Any]] = None) -> QueryFilter
Source code in apogee_core/presentation/http/helpers/query_filter_builder.py
Python
@staticmethod
def build(
    *,
    page: int = 1,
    per_page: int = 20,
    q: Optional[str] = None,
    sort: Optional[str] = None,
    fields: Optional[str] = None,
    search_fields: frozenset[str] = frozenset(),
    raw_params: Optional[dict[str, Any]] = None,
) -> QueryFilter:
    filters = _parse_bracket_filters(raw_params or {})
    return QueryFilter.from_params(
        page=page, per_page=per_page, search=q,
        search_fields=tuple(search_fields) if q else (),
        sort=sort, fields=fields, filters=filters,
    )

from_fastapi_request staticmethod

Python
from_fastapi_request(request: Any, search_fields: frozenset[str] = frozenset()) -> QueryFilter
Source code in apogee_core/presentation/http/helpers/query_filter_builder.py
Python
@staticmethod
def from_fastapi_request(
    request: Any,
    search_fields: frozenset[str] = frozenset(),
) -> QueryFilter:
    raw = dict(request.query_params)
    q = raw.pop("q", None) or None
    sort = raw.pop("sort", None) or None
    fields = raw.pop("fields", None) or None
    page = int(raw.pop("page", 1) or 1)
    per_page = int(raw.pop("per_page", 20) or 20)
    filters = _parse_bracket_filters(raw)
    return QueryFilter.from_params(
        page=page, per_page=per_page, search=q,
        search_fields=tuple(search_fields) if q else (),
        sort=sort, fields=fields, filters=filters,
    )

RedisCacheStore

Python
RedisCacheStore(client: Any | None = None, url: str = 'redis://localhost:6379/0', namespace: str = '', serializer: str = 'json')

Lazy redis-py async adapter.

JSON-encodes by default; falls back to pickle when serializer="pickle".

Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
def __init__(
    self,
    client: Any | None = None,
    url: str = "redis://localhost:6379/0",
    namespace: str = "",
    serializer: str = "json",
) -> None:
    self._url = url
    self._namespace = namespace
    self._serializer = serializer
    self._client = client

name class-attribute instance-attribute

Python
name = 'redis'

get async

Python
get(key: str) -> Any | None
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def get(self, key: str) -> Any | None:
    client = await self._ensure_client()
    raw = await client.get(self._key(key))
    return self._decode(raw)

set async

Python
set(key: str, value: Any, ttl_s: float | None = None) -> None
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def set(self, key: str, value: Any, ttl_s: float | None = None) -> None:
    client = await self._ensure_client()
    payload = self._encode(value)
    if ttl_s is not None:
        await client.set(self._key(key), payload, ex=int(ttl_s))
    else:
        await client.set(self._key(key), payload)

delete async

Python
delete(key: str) -> bool
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def delete(self, key: str) -> bool:
    client = await self._ensure_client()
    return bool(await client.delete(self._key(key)))

exists async

Python
exists(key: str) -> bool
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def exists(self, key: str) -> bool:
    client = await self._ensure_client()
    return bool(await client.exists(self._key(key)))

expire async

Python
expire(key: str, ttl_s: float) -> bool
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def expire(self, key: str, ttl_s: float) -> bool:
    client = await self._ensure_client()
    return bool(await client.expire(self._key(key), int(ttl_s)))

clear async

Python
clear() -> None
Source code in apogee_core/infrastructure/cache/redis_cache_store.py
Python
async def clear(self) -> None:
    client = await self._ensure_client()
    if not self._namespace:
        await client.flushdb()
        return
    pattern = f"{self._namespace}:*"
    async for key in client.scan_iter(pattern):
        await client.delete(key)

SqlCacheAside

Python
SqlCacheAside(*args: Any, cache: Any | None = None, cache_ttl_s: float | None = None, **kwargs: Any)

Mixin: applies cache-aside on top of a BaseSQLQueryRepository.

Subclasses pass an ICacheStore via cache=... constructor kwarg plus a cache_ttl_s (optional). The mixin keeps a per-class prefix derived from the model's name to avoid collisions across repos.

Source code in apogee_core/infrastructure/cache/sql_cache_aside.py
Python
def __init__(
    self,
    *args: Any,
    cache: Any | None = None,
    cache_ttl_s: float | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(*args, **kwargs)
    self._cache = cache
    if cache_ttl_s is not None:
        self.cache_ttl_s = cache_ttl_s
    model_name = getattr(getattr(self, "_model_class", None), "__name__", "Entity")
    self._cache_prefix = f"sql:{model_name}"

cache_ttl_s class-attribute instance-attribute

Python
cache_ttl_s: float | None = 300.0

get_by_id async

Python
get_by_id(entity_id: UUID) -> Any
Source code in apogee_core/infrastructure/cache/sql_cache_aside.py
Python
async def get_by_id(self, entity_id: UUID) -> Any:
    if self._cache is None:
        return await super().get_by_id(entity_id)  # type: ignore[misc]
    key = self._cache_key(f"id:{entity_id}")
    cached = await self._cache.get(key)
    if cached is not None:
        return cached
    value = await super().get_by_id(entity_id)  # type: ignore[misc]
    if value is not None:
        await self._cache.set(key, value, ttl_s=self.cache_ttl_s)
    return value

find_all async

Python
find_all(*args: Any, **kwargs: Any) -> Any
Source code in apogee_core/infrastructure/cache/sql_cache_aside.py
Python
async def find_all(self, *args: Any, **kwargs: Any) -> Any:
    if self._cache is None:
        return await super().find_all(*args, **kwargs)  # type: ignore[misc]
    key = self._cache_key("find:" + self._digest({"args": args, "kwargs": kwargs}))
    cached = await self._cache.get(key)
    if cached is not None:
        return cached
    value = await super().find_all(*args, **kwargs)  # type: ignore[misc]
    await self._cache.set(key, value, ttl_s=self.cache_ttl_s)
    return value

invalidate async

Python
invalidate(entity_id: UUID | None = None) -> None
Source code in apogee_core/infrastructure/cache/sql_cache_aside.py
Python
async def invalidate(self, entity_id: UUID | None = None) -> None:
    if self._cache is None:
        return
    if entity_id is not None:
        await self._cache.delete(self._cache_key(f"id:{entity_id}"))

SuccessResponse

Bases: BaseModel, Generic[T]

success class-attribute instance-attribute

Python
success: bool = True

data instance-attribute

Python
data: T

handle_domain_exception

Python
handle_domain_exception(exc: Exception) -> tuple[int, dict]

Returns (status_code, response_body) for known domain exceptions.

Source code in apogee_core/presentation/http/base_error_handlers/base_error_handler.py
Python
def handle_domain_exception(exc: Exception) -> tuple[int, dict]:
    """Returns (status_code, response_body) for known domain exceptions."""
    from apogee_core.domain.exceptions.base_exception import (
        AlreadyExistsException,
        DomainException,
        NotFoundException,
    )

    if isinstance(exc, NotFoundException):
        return 404, {"success": False, "error": {"code": exc.code, "message": exc.message}}
    if isinstance(exc, AlreadyExistsException):
        return 409, {"success": False, "error": {"code": exc.code, "message": exc.message}}
    if isinstance(exc, DomainException):
        return 400, {"success": False, "error": {"code": exc.code, "message": exc.message}}

    logger.error(f"Unhandled exception: {exc}")
    return 500, {"success": False, "error": {"code": "INTERNAL_ERROR", "message": "Internal Server Error"}}

parse_validation_errors

Python
parse_validation_errors(errors: list[dict[str, Any]]) -> list[dict[str, str]]

Converts Pydantic validation errors to API-friendly format.

Source code in apogee_core/presentation/http/helpers/validation_error_parser.py
Python
def parse_validation_errors(errors: list[dict[str, Any]]) -> list[dict[str, str]]:
    """Converts Pydantic validation errors to API-friendly format."""
    result = []
    for err in errors:
        loc = ".".join(str(p) for p in err.get("loc", []))
        result.append({
            "field": loc,
            "message": err.get("msg", "Invalid value"),
            "type": err.get("type", "value_error"),
        })
    return result

Other · Enums

DbEngineEnum

Bases: str, Enum

POSTGRES class-attribute instance-attribute

Python
POSTGRES = 'postgres'

MYSQL class-attribute instance-attribute

Python
MYSQL = 'mysql'

MONGO class-attribute instance-attribute

Python
MONGO = 'mongo'

Other · Exceptions

AlreadyExistsException

Python
AlreadyExistsException(entity: str, identifier: str)

Bases: DomainException

Source code in apogee_core/domain/exceptions/base_exception.py
Python
def __init__(self, entity: str, identifier: str) -> None:
    super().__init__(f"{entity} already exists: {identifier}", code="ALREADY_EXISTS")

DomainException

Python
DomainException(message: str, code: str = 'DOMAIN_ERROR')

Bases: Exception

Source code in apogee_core/domain/exceptions/base_exception.py
Python
def __init__(self, message: str, code: str = "DOMAIN_ERROR") -> None:
    self.message = message
    self.code = code
    super().__init__(message)

message instance-attribute

Python
message = message

code instance-attribute

Python
code = code

NotFoundException

Python
NotFoundException(entity: str, identifier: str)

Bases: DomainException

Source code in apogee_core/domain/exceptions/base_exception.py
Python
def __init__(self, entity: str, identifier: str) -> None:
    super().__init__(f"{entity} not found: {identifier}", code="NOT_FOUND")

Other · Protocols (ports)

ICacheStore

Bases: Protocol

name instance-attribute

Python
name: str

get async

Python
get(key: str) -> Any | None
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def get(self, key: str) -> Any | None: ...

set async

Python
set(key: str, value: Any, ttl_s: float | None = None) -> None
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def set(self, key: str, value: Any, ttl_s: float | None = None) -> None: ...

delete async

Python
delete(key: str) -> bool
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def delete(self, key: str) -> bool: ...

exists async

Python
exists(key: str) -> bool
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def exists(self, key: str) -> bool: ...

expire async

Python
expire(key: str, ttl_s: float) -> bool
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def expire(self, key: str, ttl_s: float) -> bool: ...

clear async

Python
clear() -> None
Source code in apogee_core/domain/services/i_cache_store.py
Python
async def clear(self) -> None: ...

ICommandRepository

Bases: Protocol[T]

create async

Python
create(entity: T) -> T
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def create(self, entity: T) -> T: ...

update async

Python
update(entity: T) -> T
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def update(self, entity: T) -> T: ...

delete async

Python
delete(entity_id: UUID) -> bool
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def delete(self, entity_id: UUID) -> bool: ...

IFilterableQueryRepository

Bases: IQueryRepository[T], Protocol[T]

find_all async

Python
find_all(filters: dict[str, Any] | None = None, page: int = 1, per_page: int = 20) -> tuple[list[T], int]
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def find_all(self, filters: dict[str, Any] | None = None, page: int = 1, per_page: int = 20) -> tuple[list[T], int]: ...

IGenericCrudService

Bases: Protocol, Generic[TCreateDTO, TUpdateDTO, TOutputDTO]

create async

Python
create(dto: TCreateDTO) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def create(self, dto: TCreateDTO) -> TOutputDTO: ...

get async

Python
get(entity_id: str) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def get(self, entity_id: str) -> TOutputDTO: ...

list async

Python
list(query: QueryFilter) -> PagedResult[TOutputDTO]
Source code in apogee_core/application/services/generic_service.py
Python
async def list(self, query: QueryFilter) -> PagedResult[TOutputDTO]: ...

update async

Python
update(entity_id: str, dto: TUpdateDTO) -> TOutputDTO
Source code in apogee_core/application/services/generic_service.py
Python
async def update(self, entity_id: str, dto: TUpdateDTO) -> TOutputDTO: ...

delete async

Python
delete(entity_id: str) -> bool
Source code in apogee_core/application/services/generic_service.py
Python
async def delete(self, entity_id: str) -> bool: ...

IMongoCommandRepository

Bases: Protocol[T]

create async

Python
create(entity: T) -> T
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def create(self, entity: T) -> T: ...

update async

Python
update(entity: T) -> T
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def update(self, entity: T) -> T: ...

delete async

Python
delete(entity_id: UUID) -> bool
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def delete(self, entity_id: UUID) -> bool: ...

IMongoQueryRepository

Bases: Protocol[T]

get_by_id async

Python
get_by_id(entity_id: UUID) -> T | None
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def get_by_id(self, entity_id: UUID) -> T | None: ...

find_all async

Python
find_all(query: QueryFilter | None = None) -> PagedResult[T]
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def find_all(self, query: QueryFilter | None = None) -> PagedResult[T]: ...

count async

Python
count(filters: dict[str, Any] | None = None) -> int
Source code in apogee_core/domain/repositories/i_base_mongo_repository.py
Python
async def count(self, filters: dict[str, Any] | None = None) -> int: ...

IQueryRepository

Bases: Protocol[T]

get_by_id async

Python
get_by_id(entity_id: UUID) -> T | None
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def get_by_id(self, entity_id: UUID) -> T | None: ...

get_all async

Python
get_all(filters: dict[str, Any] | None = None, page: int = 1, per_page: int = 20) -> tuple[list[T], int]
Source code in apogee_core/domain/repositories/i_base_repository.py
Python
async def get_all(self, filters: dict[str, Any] | None = None, page: int = 1, per_page: int = 20) -> tuple[list[T], int]: ...