BeanieRepository — MongoDB ODM Repository with Beanie

BeanieRepository is HexCore’s built-in async CRUD base for MongoDB, powered by Beanie ODM. It maps a typed domain entity T to a Beanie Document subclass D, handling entity_id remapping, logical deletes, and field-level transformations through optional resolver and serialiser maps.

BeanieRepository and all MongoDB symbols on this page require the hexcore[mongo] extra:

pip install "hexcore[mongo]"

If Beanie is not installed the class silently becomes an empty generic stub so non-MongoDB services import cleanly.

Class hierarchy

IBaseRepository[T]
  └── BaseBeanieRepository[T]           # no-session base (stateless)
        └── BeanieRepository[T, D]      # CRUD implementations

BeanieRepository is also exported under the backward-compatible alias BeanieODMCommonImplementationsRepo.

`BaseDocument`

hexcore.infrastructure.repositories.orms.beanie.BaseDocument Abstract Beanie Document base class that every collection document must extend. It adds standard audit fields and a unique entity_id index that bridges the MongoDB _id with your domain entity id.

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

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

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

Built-in fields

Field	Type	Default	Notes
`entity_id`	`Annotated[UUID, Indexed(unique=True)]`	required	Maps to the domain entity `id`
`created_at`	`Optional[datetime]`	`datetime.now()`	Set at creation
`updated_at`	`Optional[datetime]`	`datetime.now()`	Auto-updated via `@after_event([Save])`
`is_active`	`Optional[bool]`	`True`	Used for logical deletes

Settings

class Settings:
    is_root = True    # Enables polymorphic document storage
    use_cache = True  # Beanie query-result caching

`init_beanie_documents()`

hexcore.infrastructure.repositories.orms.beanie.utils.init_beanie_documents

async def init_beanie_documents() -> None

Discovers every concrete subclass of BaseDocument that has been imported into the current Python process, then calls beanie.init_beanie() against the MongoDB URI configured in ServerConfig.mongo_uri. Call this once at application startup, after all document modules have been imported.

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from hexcore.infrastructure.repositories.orms.beanie.utils import init_beanie_documents
import myapp.infrastructure.documents  # ensure documents are imported

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

app = FastAPI(lifespan=lifespan)

init_beanie_documents() reads LazyConfig.get_config().mongo_uri at call time, so you can override the URI via the HEXCORE_CONFIG_MODULE environment variable without touching your source code.

`BeanieRepository`

hexcore.infrastructure.repositories.implementations.BeanieRepository[T, D] Full CRUD repository that maps a domain entity T to a Beanie document D. All methods are async.

Generic parameters

Parameter	Bound	Description
`T`	`BaseEntity`	Domain entity type returned by every method
`D`	`BaseDocument`	Beanie document type used for MongoDB persistence

Abstract properties to implement

entity_cls

Type[T]

required

The domain entity class used to reconstruct entities from documents via model_construct.

document_cls

Type[D]

required

The Beanie document class used for all collection queries and writes.

not_found_exception

Type[Exception]

required

Exception class (not instance) raised when a document is not found. Receives the queried entity_id as its sole constructor argument.

Optional properties

fields_resolvers

FieldResolversType[D]

default:"{}"

A mapping of { entity_field: (document_field, async_callable) } applied when converting a document to a domain entity. Each async callable receives the full document instance and must return the resolved value.

fields_serializers

FieldSerializersType[T]

default:"{}"

A mapping of { entity_field: (document_field, callable) } applied when converting a domain entity to a document. Useful for embedding value objects or transforming complex types.

limit_offset_pagination

bool

default:"True"

When True, list_all applies .skip(offset).limit(limit) to the Beanie query. Set to False to return all documents.

Methods

get_by_id(entity_id: UUID)

Awaitable[T]

Queries the collection for { "entity_id": entity_id }. Raises not_found_exception(entity_id) if no document is found.

list_all(limit: Optional[int] = None, offset: int = 0)

Awaitable[List[T]]

Returns all documents in the collection. When limit_offset_pagination is True, applies .skip(offset).limit(limit).

query_all(query: QueryRequestDTO)

Awaitable[tuple[List[T], int]]

Executes a parameterised MongoDB query supporting full-text regex search, typed filters (EQ, NE, GT, GTE, LT, LTE, IN, NOT_IN, CONTAINS, STARTSWITH, ENDSWITH, IS_NULL), sorting, and limit/offset pagination. Returns the page of results and the total matching document count.

save(entity: T)

Awaitable[T]

Decorated with @register_entity_on_uow. Upserts the entity: if a document with matching entity_id exists, updates it; otherwise inserts a new document. Returns the re-hydrated entity. The entity.id field is mapped to document.entity_id on write and back to entity.id on read.

delete(entity: T)

Awaitable[None]

Performs a logical delete: sets document.is_active = False and calls document.save(). Documents are never physically removed.

`id` ↔ `entity_id` remapping

MongoDB documents managed by HexCore store the domain entity’s UUID in the entity_id field (not _id). HexCore handles this remapping transparently:

Direction	Mapping
Entity → Document (save)	`entity.id` → `document.entity_id`; MongoDB `_id` is auto-generated
Document → Entity (read)	`document.entity_id` → `entity.id`; `_id` is discarded

Complete example

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

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

    class Settings:
        name = "users"

Repository discovery

config.py

from hexcore.config import ServerConfig

config = ServerConfig(
    repository_discovery_paths={"myapp.infrastructure.repositories"},
)

Then call:

from hexcore.infrastructure.repositories.utils import discover_nosql_repositories

repos = discover_nosql_repositories()
# {"user": UserRepository, "product": ProductRepository, ...}

Domain

CQRS

Infrastructure

BeanieRepository — MongoDB ODM Repository with Beanie

Class hierarchy

`BaseDocument`

Built-in fields

Settings

`init_beanie_documents()`

`BeanieRepository`

Generic parameters

Abstract properties to implement

Optional properties

Methods

`id` ↔ `entity_id` remapping

Complete example

Repository discovery

Build docs developers (and LLMs) love

Domain

CQRS

Infrastructure

Documentation Index

​Class hierarchy

​BaseDocument

​Built-in fields

​Settings

​init_beanie_documents()

​BeanieRepository

​Generic parameters

​Abstract properties to implement

​Optional properties

​Methods

​id ↔ entity_id remapping

​Complete example

​Repository discovery

Build docs developers (and LLMs) love

Class hierarchy

`BaseDocument`

Built-in fields

Settings

`init_beanie_documents()`

`BeanieRepository`

Generic parameters

Abstract properties to implement

Optional properties

Methods

`id` ↔ `entity_id` remapping

Complete example

Repository discovery