Implement MongoDB Repositories with Beanie and HexCore

HexCore’s BeanieRepository (also exported as BeanieODMCommonImplementationsRepo) provides a ready-made CRUD implementation for MongoDB using the Beanie ODM. This page explains how to define a BaseDocument subclass, implement a concrete repository, register all documents at startup with init_beanie_documents, and handle complex fields with resolvers and serializers.

Prerequisites

Install HexCore with the Beanie extra:

pip install "hexcore[beanie]"

Set mongo_uri in your ServerConfig:

config.py

from hexcore.config import ServerConfig

config = ServerConfig(
    mongo_uri="mongodb://localhost:27017/mydb",
    repository_discovery_paths={
        "src.infrastructure.repositories",
    },
)

Core Classes

BaseDocument

Abstract Beanie Document base. Provides entity_id (UUID, unique index), created_at, updated_at, and is_active. All MongoDB documents must extend it.

BeanieRepository

Generic CRUD repository. Subclass it with your entity type T and document type D. Provides get_by_id, list_all, query_all, save, and delete.

Step 1 — Define the Beanie Document

BaseDocument maps entity_id to your entity’s domain id. The MongoDB document _id is an auto-generated ObjectId managed by Beanie; entity_id is the UUID that HexCore uses to query and convert records.

src/infrastructure/database/documents/user_document.py

from beanie import Indexed
from hexcore.infrastructure.repositories.orms.beanie import BaseDocument
import typing as t


class UserDocument(BaseDocument):
    name: str
    email: t.Annotated[str, Indexed(unique=True)]
    role: str = "member"

    class Settings:
        name = "users"   # MongoDB collection name

Do not declare entity_id, created_at, updated_at, or is_active in your subclass — they are already defined on BaseDocument. The @after_event([Save]) hook on BaseDocument automatically updates updated_at on every save.

Step 2 — Define the Domain Entity

src/domain/users/entities.py

from uuid import UUID
from hexcore.domain.base import BaseEntity


class User(BaseEntity):
    name: str
    email: str
    role: str

Step 3 — Implement the Repository

Subclass BeanieRepository[Entity, Document] and declare entity_cls, document_cls, and not_found_exception:

src/infrastructure/repositories/user_repository.py

from hexcore.infrastructure.repositories.implementations import BeanieRepository
from hexcore.domain.uow import IUnitOfWork

from src.domain.users.entities import User
from src.domain.users.exceptions import UserNotFoundException
from src.domain.users.repositories import IUserRepository
from src.infrastructure.database.documents.user_document import UserDocument


class UserRepository(BeanieRepository[User, UserDocument], IUserRepository):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return User

    @property
    def document_cls(self):
        return UserDocument

    @property
    def not_found_exception(self):
        return UserNotFoundException

Available Methods

get_by_id

async (entity_id: UUID) -> T

Queries MongoDB by entity_id. Raises not_found_exception when no document matches.

list_all

async (limit: int | None, offset: int) -> list[T]

Returns all documents. limit_offset_pagination defaults to True, so limit and offset are applied automatically. Set it to False on the repository class to always return every document.

query_all

async (query: QueryRequestDTO) -> tuple[list[T], int]

Executes a filtered, sorted, paginated MongoDB query. Returns (items, total_count). Supports all FilterOperator values including CONTAINS, STARTSWITH, IN, and more.

save

async (entity: T) -> T

Decorated with @register_entity_on_uow. Upserts the document (creates on first call, updates on subsequent calls matching entity_id). Returns the refreshed entity.

delete

async (entity: T) -> None

Performs a logical delete by setting is_active = False on the document. The document remains in MongoDB.

Registering Documents at Startup

Call init_beanie_documents() once during application startup. It auto-discovers every concrete BaseDocument subclass in your codebase and initialises Beanie with the MongoDB connection from ServerConfig.mongo_uri:

src/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from hexcore.infrastructure.repositories.orms.beanie.utils import init_beanie_documents


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_beanie_documents()   # registers UserDocument, OrderDocument, etc.
    yield


app = FastAPI(lifespan=lifespan)

init_beanie_documents uses get_all_concrete_subclasses(BaseDocument) internally to collect every non-abstract BaseDocument subclass that has been imported at the time it is called. Import your document modules before calling it (or rely on HexCore’s repository discovery to import them automatically).

Field Resolvers and Serializers

`fields_resolvers` — Document → Entity

Resolvers translate raw document values into richer entity types after a document is loaded. Each entry maps an entity field name to a (source_field, async_resolver_fn) tuple. The resolver receives the full document instance.

src/infrastructure/repositories/order_repository.py

from hexcore.infrastructure.repositories.decorators import cycle_protection_resolver
from hexcore.infrastructure.repositories.implementations import BeanieRepository
from hexcore.domain.uow import IUnitOfWork

from src.domain.orders.entities import Order
from src.domain.orders.enums import OrderStatus
from src.domain.orders.exceptions import OrderNotFoundException
from src.infrastructure.database.documents.order_document import OrderDocument


class OrderRepository(BeanieRepository[Order, OrderDocument]):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return Order

    @property
    def document_cls(self):
        return OrderDocument

    @property
    def not_found_exception(self):
        return OrderNotFoundException

    @property
    def fields_resolvers(self):
        @cycle_protection_resolver
        async def resolve_status(doc: OrderDocument):
            return OrderStatus(doc.status)

        return {
            # entity_field: (document_field, async_resolver_fn)
            "status": ("status", resolve_status),
        }

`fields_serializers` — Entity → Document

Serializers convert complex entity fields to MongoDB-storable primitives before the document is saved. Each entry maps an entity field to a (dest_field, sync_serializer_fn) tuple.

    @property
    def fields_serializers(self):
        return {
            # entity_field: (document_field, serializer_fn)
            "status": ("status", lambda entity: entity.status.value),
        }

fields_resolvers runs on read (Document → Entity). fields_serializers runs on write (Entity → Document). Use @cycle_protection_resolver for resolvers that load related entities to avoid infinite recursion in bidirectional relationships.

Complete End-to-End Example

Define the entity and exception

src/domain/users/entities.py

from uuid import UUID
from hexcore.domain.base import BaseEntity

class User(BaseEntity):
    name: str
    email: str
    role: str

src/domain/users/exceptions.py

class UserNotFoundException(Exception):
    def __init__(self, user_id):
        super().__init__(f"User {user_id} not found.")

Define the Beanie document

src/infrastructure/database/documents/user_document.py

import typing as t
from beanie import Indexed
from hexcore.infrastructure.repositories.orms.beanie import BaseDocument

class UserDocument(BaseDocument):
    name: str
    email: t.Annotated[str, Indexed(unique=True)]
    role: str = "member"

    class Settings:
        name = "users"

Define the domain repository interface

src/domain/users/repositories.py

import abc
from uuid import UUID
from hexcore.domain.repositories import IBaseRepository
from .entities import User

class IUserRepository(IBaseRepository[User]):
    @abc.abstractmethod
    async def get_by_email(self, email: str) -> User: ...

Implement the repository

src/infrastructure/repositories/user_repository.py

from hexcore.infrastructure.repositories.implementations import BeanieRepository
from hexcore.domain.uow import IUnitOfWork
from src.domain.users.entities import User
from src.domain.users.exceptions import UserNotFoundException
from src.domain.users.repositories import IUserRepository
from src.infrastructure.database.documents.user_document import UserDocument

class UserRepository(BeanieRepository[User, UserDocument], IUserRepository):
    def __init__(self, uow: IUnitOfWork) -> None:
        super().__init__(uow)

    @property
    def entity_cls(self):
        return User

    @property
    def document_cls(self):
        return UserDocument

    @property
    def not_found_exception(self):
        return UserNotFoundException

    async def get_by_email(self, email: str) -> User:
        doc = await UserDocument.find_one({"email": email})
        if not doc:
            raise UserNotFoundException(email)
        from hexcore.infrastructure.repositories.utils import to_entity_from_model_or_document
        return await to_entity_from_model_or_document(doc, User, is_nosql=True)

src/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from hexcore.infrastructure.repositories.orms.beanie.utils import init_beanie_documents

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_beanie_documents()
    yield

app = FastAPI(lifespan=lifespan)

src/application/use_cases/create_user.py

from uuid import uuid4
from src.domain.users.entities import User
from src.domain.users.repositories import IUserRepository
from hexcore.domain.uow import IUnitOfWork

class CreateUserUseCase:
    def __init__(self, repo: IUserRepository, uow: IUnitOfWork) -> None:
        self.repo = repo
        self.uow = uow

    async def execute(self, name: str, email: str) -> User:
        user = User(id=uuid4(), name=name, email=email, role="member")
        async with self.uow:
            saved = await self.repo.save(user)
            await self.uow.commit()
        return saved

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Implement MongoDB Repositories with Beanie and HexCore

Prerequisites

Core Classes

BaseDocument

BeanieRepository

Step 1 — Define the Beanie Document

Step 2 — Define the Domain Entity

Step 3 — Implement the Repository

Available Methods

Registering Documents at Startup

Field Resolvers and Serializers

`fields_resolvers` — Document → Entity

`fields_serializers` — Entity → Document

Complete End-to-End Example

Build docs developers (and LLMs) love

Get Started

Core Concepts

Configuration

Domain Layer

CQRS & Messaging

Infrastructure

CLI Reference

Documentation Index

​Prerequisites

​Core Classes

BaseDocument

BeanieRepository

​Step 1 — Define the Beanie Document

​Step 2 — Define the Domain Entity

​Step 3 — Implement the Repository

​Available Methods

​Registering Documents at Startup

​Field Resolvers and Serializers

​fields_resolvers — Document → Entity

​fields_serializers — Entity → Document

​Complete End-to-End Example

Build docs developers (and LLMs) love

Prerequisites

Core Classes

Step 1 — Define the Beanie Document

Step 2 — Define the Domain Entity

Step 3 — Implement the Repository

Available Methods

Registering Documents at Startup

Field Resolvers and Serializers

`fields_resolvers` — Document → Entity

`fields_serializers` — Entity → Document

Complete End-to-End Example