MongoDB Backend

Therismos provides MongoDB visitors for expressions, sorting, and grouping. Compatible with PyMongo and Motor.

Installation

pip install therismos[mongodb]        # synchronous PyMongo
pip install therismos[mongodb-async]  # asynchronous Motor

Expression Filtering

from therismos import F, optimize
from therismos.expr.visitors.mongo import MongoVisitor

age = F("age")
status = F("status")
country = F("country")

expr = (age >= 21) & (status == "active") & (country.is_in("US", "UK", "CA"))
optimized, _ = optimize(expr)

visitor = MongoVisitor()
mongo_filter = optimized.accept(visitor)
# {"age": {"$gte": 21}, "status": "active", "country": {"$in": ["US", "UK", "CA"]}}

Expression Mapping

import re
from therismos import F, TRUE, FALSE
from therismos.expr.visitors.mongo import MongoVisitor

visitor = MongoVisitor()
email = F("email")
age = F("age")
status = F("status")
name = F("name")

# Regex (case-insensitive)
email.matches(r".*@example\.com$", re.IGNORECASE).accept(visitor)
# {"email": {"$regex": ".*@example\\.com$", "$options": "i"}}

# Range
((age >= 18) & (age <= 65)).accept(visitor)
# {"age": {"$gte": 18, "$lte": 65}}

# NOT
(~(age < 18)).accept(visitor)
# {"$nor": [{"age": {"$lt": 18}}]}

# Null check
name.is_not_null().accept(visitor)
# {"name": {"$ne": null}}

# Constants
TRUE.accept(visitor)   # {}
FALSE.accept(visitor)  # {"$expr": false}

AND Optimization

# Default: merge simple AND fields
visitor = MongoVisitor(optimize_simple_and=True)
# {"age": {"$gt": 18}, "name": "Alice"}

# Disable: always use $and
visitor = MongoVisitor(optimize_simple_and=False)
# {"$and": [{"age": {"$gt": 18}}, {"name": "Alice"}]}

Using with PyMongo

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
results = client["mydb"]["users"].find(mongo_filter)

Using with Motor (async)

from motor.motor_asyncio import AsyncIOMotorClient

async def find_users():
    client = AsyncIOMotorClient("mongodb://localhost:27017/")
    cursor = client["mydb"]["users"].find(mongo_filter)
    return await cursor.to_list(length=100)

Sorting

from therismos.sorting import SortSpec, SortCriterion, SortOrder
from therismos.sorting.visitors.mongo import MongoVisitor

spec = SortSpec([
    SortCriterion("created_at", SortOrder.DESCENDING),
    SortCriterion("name", SortOrder.ASCENDING),
])

mongo_sort = spec.accept(MongoVisitor())
# {"created_at": -1, "name": 1}

# Use with PyMongo
# cursor = collection.find().sort(list(mongo_sort.items()))

Grouping

from therismos.grouping import GroupSpec, Aggregation, AggregationFunction
from therismos.grouping.visitors.mongo import MongoVisitor

spec = GroupSpec(
    group_by=["category", "region"],
    aggregations=[
        Aggregation("total", AggregationFunction.COUNT),
        Aggregation("avg_price", AggregationFunction.AVERAGE, "price"),
        Aggregation("p95_latency", AggregationFunction.P95, "latency"),
    ],
)

group_stage = spec.accept(MongoVisitor())
# {
#     "$group": {
#         "_id": {"category": "$category", "region": "$region"},
#         "total": {"$sum": 1},
#         "avg_price": {"$avg": "$price"},
#         "p95_latency": {"$percentile": {"input": "$latency", "p": [0.95], "method": "approximate"}}
#     }
# }

# Pipeline usage
pipeline = [{"$match": filter_stage}, group_stage]
results = collection.aggregate(pipeline)

Single Field Simplification

spec = GroupSpec(group_by=["status"], aggregations=[Aggregation("count", AggregationFunction.COUNT)])

MongoVisitor().accept(spec)
# {"$group": {"_id": "$status", "count": {"$sum": 1}}}

MongoVisitor(simplify_single_group=False).accept(spec)
# {"$group": {"_id": {"status": "$status"}, "count": {"$sum": 1}}}

Note: Percentile functions (MEDIAN, Q1, Q3, P01–P99) require MongoDB 7.0+.