CRUD, search, and pagination
Create records, perform unique lookups, filter and search collections, aggregate values, and paginate results.
Work through each step in order. The examples use the generated LenAPI project structure and are designed to be adapted to your app.
Create and save a record
Use Model.create(...) for synchronous code; it inserts, commits, refreshes, and returns the model instance. For async routes, await Model.acreate(...). You can also instantiate directly, then call save() or await asave() after changing fields.
# Sync
student = Student.create(person_name="Ada", course="Python")
student.course = "Databases"
student.save()
# Async
student = await Student.acreate(person_name="Ada", course="Python")
student.course = "Databases"
await student.asave()Get all rows or one row
For a synchronous list use Model.get.all(); the matching async manager call is await Model.aget.all(). For a filtered async list use await Model.afilter(...).all(). To fetch one row, use Model.get(id=...).get() or await Model.aget(id=...). Single-row lookups return None when no row matches; use the sync *_or_404 helper or raise HTTPException in an async route.
# All rows: full model instances
students = Student.get.all()
async_students = await Student.aget.all()
# One row by id
student = Student.get(id=1).get()
async_student = await Student.aget(id=1)
# Check whether a row exists
exists = Student.get(id=1).exists()
async_exists = await Student.aget(id=1).exists()Fetch related rows through foreign keys and ManyToMany
Assuming Student.classroom_id is a ForeignKeyField and Teacher.courses is a ManyToManyField, sync code can resolve the foreign key with .special('classroom_id') and read a collection with teacher.courses.all(). In async code, fetch the foreign-key target with a second awaited lookup. For async ManyToMany, eager-load the collection with SQLAlchemy selectinload and an injected AsyncSession before accessing it; direct lazy relationship access cannot perform async I/O safely.
# Sync foreign key: special() resolves classroom_id to a Classroom object.
student = Student.get(id=1).special("classroom_id")
classroom = student.classroom_id # Classroom instance, or None
classroom_name = classroom.name if classroom else None
# Async foreign key: follow the stored id with an async lookup.
student = await Student.aget(id=1)
classroom = None
if student is not None and student.classroom_id is not None:
classroom = await Classroom.aget(id=student.classroom_id)
# Sync ManyToMany collection
teacher = Teacher.get(id=1).get()
courses = teacher.courses.all()
# Async ManyToMany: eager-load before reading the collection.
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@app.get("/teachers/{teacher_id}/courses")
async def teacher_courses(teacher_id: int, db: AsyncSession):
result = await db.scalars(
select(Teacher)
.options(selectinload(Teacher.courses))
.where(Teacher.id == teacher_id)
)
teacher = result.one_or_none()
if teacher is None:
return []
return [{"id": course.id, "name": course.name}
for course in teacher.courses.all()]Declare both relationships in app/models.py and run python lenora.py migration. The async example requires ACTIVATE_ASYNC = True and an installed async database driver.
Select only the columns you need
Pass column names to Model.get(...), then give lookup filters to .get(...). One selected column returns a scalar; multiple selected columns return a mapping. For async access, select with afilter().values(...) and await .get(...). This only projects query results; it does not change the model or database schema.
# Sync: one column returns a scalar
name = Student.get("person_name").get(id=1)
# Sync: multiple columns return a mapping
summary = Student.get("person_name", "course").get(id=1)
# Async projection
async_summary = await Student.afilter().values(
"person_name", "course"
).get(id=1)Filter, search, and order
filter(**conditions) builds a query and all() executes it. Supported suffixes are exact, iexact, ne, gt, gte, lt, lte, in, notin, contains, icontains, startswith, istartswith, endswith, iendswith, and isnull. Chain values(), order(), limit(), and offset() before execution. Use afilter() and await the terminal operation in async code. search(...) matches every supplied search term; search_any(query, *fields) matches the term in any listed string field.
# Sync: exact and lookup filters, projection and descending order
students = Student.filter(
course__icontains="py",
id__gte=1,
).values("person_name", "course").order("-id").all()
# Async equivalent
async_students = await Student.afilter(
course__icontains="py",
id__gte=1,
).values("person_name", "course").order("-id").all()
# Search all terms, or search one term across fields (OR)
matched = Student.search(person_name="ada", course="python").all()
any_field = Student.search_any("python", "person_name", "course").all()
async_matched = await Student.asearch(person_name="ada").all()
async_any = await Student.asearch_any(
"python", "person_name", "course"
).all()Paginate and return metadata
Pass non-negative limit and offset values to all(), or chain limit() and offset() first. The returned list-like result also exposes total_count, page_count, total_pages, and pagination metadata. These examples declare StudentOutput and a page response schema, then use response_model so Swagger documents the same response shape for sync and async routes.
from typing import Any
from pydantic import BaseModel, ConfigDict
from lenapi import LENAPI
from app.models import Student
app = LENAPI()
class StudentOutput(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
person_name: str | None
course: str | None
class StudentPageOutput(BaseModel):
results: list[StudentOutput]
pagination: dict[str, Any] | None
# Sync route
@app.get("/students", response_model=StudentPageOutput)
def list_students(limit: int = 20, offset: int = 0):
page = Student.filter(course="Python").all(limit=limit, offset=offset)
return {
"results": [StudentOutput.model_validate(row).model_dump(mode="json")
for row in page],
"pagination": page.pagination,
}
# Async route
@app.get("/async/students", response_model=StudentPageOutput)
async def list_students_async(limit: int = 20, offset: int = 0):
page = await Student.afilter(course="Python").all(
limit=limit, offset=offset
)
return {
"results": [StudentOutput.model_validate(row).model_dump(mode="json")
for row in page],
"pagination": page.pagination,
}Aggregate data
Model.count() counts every row; Model.filter(...).count() counts matching rows. Model.aggregate() and filtered-query aggregate() run Count, Sum, Min, Max, or Avg in the database. Name each aggregate result with a keyword. Pass group_by to return one result dictionary per group. Decimal sums and averages return Decimal values.
from lenapi import model
class Purchase(model.Model):
user_id = model.IntegerField(nullable=False)
amount = model.DecimalField(max_digits=12, decimal_places=2)
status = model.StringField(max_length=20)
# Row counts
total_purchases = Purchase.count()
paid_count = Purchase.filter(status="paid").count()
# One aggregate dictionary for all matching rows
spend = Purchase.filter(status="paid").aggregate(
total=model.Sum("amount"),
lowest=model.Min("amount"),
highest=model.Max("amount"),
average=model.Avg("amount"),
purchases=model.Count(),
)
# One dictionary per user
by_user = Purchase.filter(status="paid").aggregate(
group_by="user_id",
total=model.Sum("amount"),
purchases=model.Count(),
)Delete with intent
Delete a unique record with a get query or delete a filtered set. Model.all().delete() drops a whole table, so use it only when that destructive behavior is explicitly intended.
Student.get(person_name="Ada").delete()
Student.filter(course="Old Course").delete()