DATABASE GUIDE · DOCUMENT CRUD8 min read

MongoDB CRUD (beta)

Configure MongoDB and create, read, update, and delete documents with LenAPI's separate MongoDocument API.

IN THIS GUIDE

Work through each step in order. The examples use the generated LenAPI project structure and are designed to be adapted to your app.

Understand the beta API and install its driver

MongoDB is currently a beta backend. It uses MongoDocument and MongoDatabase, not SQL model.Model. SQL relationships, SQL query helpers, and Alembic schema revisions do not apply to MongoDB documents. Install LenAPI's MongoDB extra, which includes the official PyMongo driver.

Python

python -m pip install 'lenapi[mongodb]'

TIP

Use this guide to test document CRUD and provide feedback. MongoDB document fields are schema-flexible; changing a Python field does not run an Alembic migration.

Select MongoDB in settings.py

Choose MongoDB as the project's one active database and include the database name in the URI path. ACTIVATE_ASYNC controls whether LenAPI also creates the asynchronous MongoDB client. Store production credentials in your deployment secret manager or environment variables, not in committed source.

Python

# settings.py
DATABASE_TYPE = "mongodb"
DATABASE_URL = "mongodb://127.0.0.1:27017/school"
ACTIVATE_ASYNC = False  # Set True if routes will use async MongoDocument methods

TIP

For a managed MongoDB service, use its connection URI and URL-encode reserved characters in credentials. If the URI has no database path, configure MONGODB_DATABASE.

Define and bind a MongoDocument

A MongoDocument subclass is a Pydantic document model. Set collection_name to choose its collection; otherwise LenAPI uses the lower-case class name. Bind the class to the MongoDatabase created by LENAPI before serving requests. MongoDocument supplies an optional string id mapped to MongoDB's _id.

Python

# app/models.py
from lenapi import MongoDocument

class Student(MongoDocument):
    collection_name = "students"
    name: str
    course: str

# main.py
from lenapi import LENAPI
from app.models import Student

app = LENAPI()
Student.bind(app.mongodb)

Implement synchronous create, read, update, and delete routes

Use Student.create() to validate and insert a document, Student.get() for one exact-match lookup, and Student.filter() for exact-match lists. Change fields on the returned document and call save() to update it; call delete() to remove it. MongoDB IDs are strings, so declare the path id as str.

Python

from pydantic import BaseModel
from lenapi import HTTPException
from app.models import Student

class StudentInput(BaseModel):
    name: str
    course: str

@app.post("/students")
def create_student(data: StudentInput):
    student = Student.create(**data.model_dump())
    return student.model_dump(mode="json")

@app.get("/students")
def list_students(course: str | None = None):
    students = Student.filter(course=course) if course else Student.all()
    return [student.model_dump(mode="json") for student in students]

@app.get("/students/{student_id}")
def get_student(student_id: str):
    student = Student.get(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    return student.model_dump(mode="json")

@app.put("/students/{student_id}")
def update_student(student_id: str, data: StudentInput):
    student = Student.get(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    student.name = data.name
    student.course = data.course
    student.save()
    return student.model_dump(mode="json")

@app.delete("/students/{student_id}")
def delete_student(student_id: str):
    student = Student.get(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    student.delete()
    return {"deleted": True, "id": student_id}

TIP

filter() and all() currently return the matching documents as a Python list; avoid loading an unbounded collection in a large production dataset. Use an index for fields queried frequently.

Use the async document methods

Set ACTIVATE_ASYNC = True before constructing LENAPI, then await MongoDocument's async methods. The method names are acreate, aget, afilter, aall, asave, and adelete. Do not call the synchronous variants from an async handler when you want non-blocking MongoDB I/O.

Python

# settings.py
ACTIVATE_ASYNC = True

# Async routes
@app.post("/async/students")
async def create_student_async(data: StudentInput):
    student = await Student.acreate(**data.model_dump())
    return student.model_dump(mode="json")

@app.get("/async/students/{student_id}")
async def get_student_async(student_id: str):
    student = await Student.aget(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    return student.model_dump(mode="json")

@app.get("/async/students")
async def list_students_async(course: str | None = None):
    students = await Student.afilter(course=course) if course else await Student.aall()
    return [student.model_dump(mode="json") for student in students]

@app.put("/async/students/{student_id}")
async def update_student_async(student_id: str, data: StudentInput):
    student = await Student.aget(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    student.name = data.name
    student.course = data.course
    await student.asave()
    return student.model_dump(mode="json")

@app.delete("/async/students/{student_id}")
async def delete_student_async(student_id: str):
    student = await Student.aget(id=student_id)
    if student is None:
        raise HTTPException(404, "Student not found")
    await student.adelete()
    return {"deleted": True, "id": student_id}

Filter documents, add indexes, and know the limits

MongoDocument.get() and filter() accept exact field-value matches, including id. Use count() for counts and ensure_index() for indexes. The current filter API does not expose SQL-style operators, relationship traversal, or paginated query results. MongoDB collections are created on first write; lenora.py migration or migrate can initialize imported document collections and framework state, but they do not generate Alembic revisions for field changes.

Python

# Exact match and count
student = Student.get(course="Python")
python_count = Student.count(course="Python")

# Create a unique index once during setup
Student.ensure_index("email", unique=True)

# Async equivalents
student = await Student.aget(course="Python")
python_count = await Student.acount(course="Python")
await Student.aensure_index("email", unique=True)

# Initialize imported MongoDocument collections
python lenora.py migration
python lenora.py migrate

TIP

Switching DATABASE_TYPE from SQL to MongoDB does not copy or convert SQL tables and rows. Define and bind MongoDocument classes, then migrate data separately if needed. Review the beta status before choosing MongoDB for production.