varco-beanie 1.0.6

Beanie (Motor / MongoDB) async backend for varco — runtime ORM generation, repository, and UoW

varco-beanie

Beanie (Motor / MongoDB) async backend for varco.

Generates Beanie Document classes at runtime from your DomainModel subclasses — no hand-written Document models needed. Requires varco-core.

---

Install

pip install varco-beanie

Requirements

• Python ≥ 3.12
• MongoDB ≥ 4.0
• For multi-document transactions: a MongoDB replica set or sharded cluster

---

Features

• Zero-boilerplate ODM — BeanieModelFactory generates Document subclasses at runtime from your DomainModel classes; no duplication
• Full repository — AsyncBeanieRepository implements AsyncRepository (CRUD, exists(), stream_by_query())
• Unit of Work — BeanieUnitOfWork wraps Motor session lifecycle with optional transactions
• One-liner bootstrap — BeanieRepositoryProvider + BeanieFastrestApp wire everything including init_beanie()
• Query integration — accepts varco-core QueryParams / QueryBuilder AST natively

---

What's in the package

Module	Purpose
factory.py	BeanieModelFactory — generates Document subclasses; BeanieDocRegistry — escape hatch to access the generated Document
repository.py	AsyncBeanieRepository — Motor-backed CRUD + exists() + stream_by_query()
uow.py	BeanieUnitOfWork — Motor session lifecycle (optional transactions)
provider.py	BeanieRepositoryProvider — wires factory + repos + UoW + init_beanie()
bootstrap.py	BeanieConfig, BeanieFastrestApp — one-liner app setup

---

Quick start

Bootstrap (one-liner)

from motor.motor_asyncio import AsyncIOMotorClient
from varco_beanie import BeanieConfig, BeanieFastrestApp

client = AsyncIOMotorClient("mongodb://localhost:27017")

app = BeanieFastrestApp(BeanieConfig(
    motor_client=client,
    db_name="myapp",
    entity_classes=(User, Post),
))

await app.init()                    # calls beanie.init_beanie() internally
uow_provider = app.uow_provider     # ready to inject as IUoWProvider

Manual setup

from varco_beanie import BeanieRepositoryProvider

provider = BeanieRepositoryProvider(motor_client=client, db_name="myapp")
provider.register(User, Post)
await provider.init()

async with provider.make_uow() as uow:
    user = await uow.users.save(User(name="Edo", email="[e-mail]"))
    print(user.pk)

Transactions (replica set required)

provider = BeanieRepositoryProvider(
    motor_client=client,
    db_name="myapp",
    transactional=True,   # wraps each UoW in a Motor session transaction
)

Query integration

from varco_core import QueryBuilder, QueryParams

async with provider.make_uow() as uow:
    # exists() — uses .count(), no document load
    if await uow.posts.exists(post_id):
        ...

    # stream_by_query() — Motor batches internally, bounded memory
    params = QueryParams(node=QueryBuilder().eq("published", True).build())
    async for post in uow.posts.stream_by_query(params):
        await process(post)

Access the generated Beanie Document (escape hatch)

from varco_beanie import BeanieDocRegistry

PostDoc = BeanieDocRegistry.get(Post)
# use PostDoc for Beanie-specific operations not exposed by the repository

---

Notes

• CheckConstraint entries in varco-core metadata are silently ignored — MongoDB has no SQL CHECK constraints. Use Pydantic validators instead.
• Foreign key hints are metadata only — MongoDB has no FK enforcement.
• Composite PKs are emulated via compound unique indexes; find_by_id(pk_tuple) issues a find_one with a composite filter.

---

Related packages

Package	Description
varco-core	Domain model, service layer, query AST, JWT — required dependency
varco-sa	SQLAlchemy async backend (alternative to this package)

---

Links

• Repository: https://github.com/edoardoscarpaci/varco
• Full docs: https://github.com/edoardoscarpaci/varco#beanie-backend
• Issue tracker: https://github.com/edoardoscarpaci/varco/issues