mongo-aggro 0.2.2

MongoDB Aggregation Pipeline Builder with Pydantic

Mongo Aggro

A Python package for building MongoDB aggregation pipelines with Pydantic.

Features

• Type-safe - Pydantic models with full type hints
• Zero-conversion - Pass Pipeline directly to collection.aggregate()
• Python operators - Use F("age") > 18 instead of {"$gt": ["$age", 18]}
• 149 expression operators - Arithmetic, string, date, array, and more
• 46 aggregation stages - All major MongoDB stages supported
• 31 query predicates - Comparison, logical, geospatial, bitwise
• 17 accumulators - Sum, Avg, Min, Max, Push, TopN, etc.

Installation

uv add mongo-aggro
# or
pip install mongo-aggro

Quick Start

from mongo_aggro import Pipeline, Match, Group, Sort, Expr
from mongo_aggro.expressions import F

# Traditional approach
pipeline = Pipeline([
    Match(query={"status": "active"}),
    Group(id="$category", accumulators={"total": {"$sum": "$amount"}}),
    Sort(fields={"total": -1})
])

# Python operator syntax
pipeline = Pipeline([
    Match(query=Expr(expression=(F("status") == "active")).model_dump()),
    Group(id="$category", accumulators={"total": {"$sum": "$amount"}}),
])

# Pass directly to MongoDB
results = collection.aggregate(pipeline)

Package Structure

# Import from root (most common)
from mongo_aggro import Pipeline, Match, Group, Sort, Expr

# Import expressions from sub-package
from mongo_aggro.expressions import F, AddExpr, ConcatExpr, CondExpr

# Import stages by category
from mongo_aggro.stages import Match, Project, Lookup, SetWindowFields

# Import query operators
from mongo_aggro.operators import In, Regex, GeoWithin, Exists

Expression Operators with Python Syntax

from mongo_aggro import Expr, Match
from mongo_aggro.expressions import F

# Comparison operators
F("age") > 18        # {"$gt": ["$age", 18]}
F("status") == "active"  # {"$eq": ["$status", "active"]}

# Logical operators (use & | ~ instead of and or not)
(F("age") >= 18) & (F("status") == "active")
(F("type") == "premium") | (F("balance") > 1000)

# Use in Match stage
Match(query=Expr(expression=(
    (F("status") == "active") & (F("age") > 18)
)).model_dump())

Stages

Category	Stages
Core	Match, Project, Group, Sort, Limit, Skip, Unwind, AddFields, Set, Unset
Join	Lookup, GraphLookup, UnionWith
Aggregation	Bucket, BucketAuto, Facet, SortByCount, Count
Output	Out, Merge
Window	SetWindowFields, Densify, Fill
Geospatial	GeoNear
Search	Search, SearchMeta, VectorSearch, RankFusion
Change Stream	ChangeStream, ChangeStreamSplitLargeEvent
Admin	CollStats, IndexStats, CurrentOp, ListSessions

Expression Operators

Category	Examples
Comparison	Eq, Ne, Gt, Gte, Lt, Lte, Cmp
Logical	And, Or, Not
Arithmetic	Add, Subtract, Multiply, Divide, Mod, Abs, Ceil, Floor
String	Concat, Split, ToLower, ToUpper, Trim, RegexMatch
Array	ArraySize, Filter, Map, Reduce, Slice, InArray
Date	DateAdd, DateDiff, DateToString, Year, Month, Day
Type	ToDate, ToString, ToInt, Convert, Type
Conditional	Cond, IfNull, Switch
Set	SetUnion, SetIntersection, SetDifference
Window	Rank, DenseRank, Shift, ExpMovingAvg

Query Predicates

from mongo_aggro import Match, In, Regex, Exists, GeoWithin

Match(query={
    "status": In(values=["active", "pending"]).model_dump(),
    "email": Regex(pattern="@company\\.com$").model_dump(),
    "profile": Exists(exists=True).model_dump(),
})

Accumulators

from mongo_aggro import Group, Sum, Avg, Max, Push, merge_accumulators

Group(
    id="$category",
    accumulators=merge_accumulators(
        Sum(name="total", field="amount"),
        Avg(name="average", field="price"),
        Max(name="highest", field="price"),
    )
)

Documentation

• User Guide
• API Reference
• Examples

License

MIT