Skip to content

Best Practices

Team-wide conventions for building fast, testable MongoDB services with mongo-ops.

🏗️ Layering

  1. One repository per collection. Encapsulate every query the domain needs behind repository methods; keep Mongo details ($regex, $inc, projections) inside the repository.
  2. Keep models thin. BaseDocument for the shape; use Pydantic Field constraints for validation; never put business rules in the model.
  3. Use services for cross-repository logic. A Service composes multiple repositories (and TransactionManager) — routes stay thin.
  4. Expose get_many(filter=..., skip=..., limit=..., sort=...) instead of raw find for list endpoints — you get controlled pagination for free.

🔄 Lifecycle

  1. Connect once, in the lifespan. MongoConnectionManager.lifespan(...) (or explicit connect/disconnect) — never lazily per request.
  2. Construct repositories after connect(). Module-level Repo() before connection raises RuntimeError("Database not connected..."). Use dependencies or construct inside the lifespan/request.
  3. Order the cache lifecycle strictly: set_cache_backend(backend) → initialize_cache() (after connect, before use) → shutdown_cache() on exit.
  4. Register all models up front via ModelRegistry.register(...) and let initialize_all() create indexes once at startup (idempotent).

🗄️ Data & Performance

  1. Declare indexes for every hot query. Single-field, composite, and optioned (unique/TTL) specs all work via ModelRegistry.register — see use case 13.
  2. Cache only hot, low-write _id reads. Use CachedBaseRepository for lookups-by-id; invalidate (update/delete handle it) and pick a sensible default_ttl.
  3. Populate at the repository boundary. PopulatingRepository resolves refs on read and depopulates on write; do not hand-roll joins in endpoints.
  4. Respect the populate invariants: rules name the field that holds the reference and that is replaced; patch() cannot touch FK fields — use update().

🔁 Transactions & Errors

  1. Use transactions for multi-document writes. TransactionManager.start_session (inline) or execute_transaction (list of ops) — and pass session= to every collection call inside.
  2. Handle the library's real exceptions at the edges: DuplicateKeyError → 409, InvalidId → 400, CircularReferenceError → 409, ValueError guides → 400/422 (see Error Handling).

🧪 Testing

  1. Default to mock-based unit tests. Patch MongoConnectionManager.get_database, use AsyncMock collections and cursor chains — the whole suite runs without MongoDB (use case 14).
  2. Mirror the library tests. tests/test_{repository,populating_repository,cache,registry,transactions}.py are canonical examples of every pattern above.
  3. Use type hints end-to-end — mypy-gated CI (see pyproject) catches drift early.