Pipeline¶

@dataclass
class Pipeline(AsAlias)

Pipeline(*, steps: list[Step] = [])

def match(self, *predicates: MongoQuery | Predicate) -> Self

# Raw dict
Pipeline().match({"status": "active"})

# Predicate
Pipeline().match(Field("age").gte(18))

# Multiple predicates (AND)
Pipeline().match(Field("status").eq("active"), Field("age").gte(18))

# Combined with &
Pipeline().match(Field("status").eq("active") & Field("age").gte(18))

def sort(self, *spec: SortPayload) -> Self

# String field name (ascending by default)
Pipeline().sort("name")

# AttributeSpec tokens
Pipeline().sort(User.age.desc(), User.name.asc())

# Dict
Pipeline().sort({"age": -1, "name": 1})

# List of tokens
Pipeline().sort([User.age.desc(), User.name.asc()])

def skip(self, size: PositiveInteger | None, /) -> Self

Pipeline().skip(20)

def take(self, size: PositiveInteger | None, /) -> Self

Pipeline().skip(20).take(10)

def project(self, *projection: type[Model] | list[Aliased] | Mapping | Aliased) -> Self

# Project to a Model (includes only that model's fields)
Pipeline().project(UserSummary)

# Project with dict
Pipeline().project({"name": True, "age": True})

# Project with Field keep/remove
Pipeline().project(
    Field("name").keep(),
    Field("age").keep(alias="person_age"),
    Field("internal").remove(),
)

# Project with list
Pipeline().project([Field("name").keep(), Field("age").keep()])

def group(self, *accumulators: Aliased | dict | list | None, by: AnyExpression) -> Self

from gault.accumulators import Sum, Avg, Count

# Dict form
Pipeline().group(
    {"total": Sum("$amount"), "avg": Avg("$score")},
    by="$category",
)

# Spread Aliased form
Pipeline().group(
    Sum("$amount").alias("total"),
    Avg("$score").alias("avg"),
    by="$category",
)

# Group all documents
Pipeline().group({"count": Count()}, by=None)

def set(self, *fields: Mapping | list[Aliased] | Aliased) -> Self

# Dict form
Pipeline().set({"total": {"$multiply": ["$price", "$qty"]}, "status": "done"})

# Spread Aliased form
Pipeline().set(
    Field("total").assign({"$multiply": ["$price", "$qty"]}),
    Field("status").assign("done"),
)

def set_field(self, field: FieldLike, value: AnyExpression, /) -> Self

Pipeline().set_field("status", "processed")

def unset(self, *fields: FieldLike) -> Self

Pipeline().unset("_id", "internal", "temp")

def unwind(
    self,
    field: FieldLike,
    /,
    *,
    include_array_index: str | None = None,
    preserve_null_and_empty_arrays: bool | None = None,
) -> Self

Pipeline().unwind("$items", include_array_index="idx", preserve_null_and_empty_arrays=True)

def bucket(
    self,
    *output: Aliased | dict | list | None,
    by: AnyExpression,
    boundaries: list[T],
    default: str | None = None,
) -> Self

Pipeline().bucket(
    {"count": Sum(1)},
    by="$age",
    boundaries=[0, 18, 65, 100],
    default="other",
)

def bucket_auto(
    self,
    *output: Aliased | dict | list | None,
    by: AnyExpression,
    buckets: int,
    granularity: Granularity | None = None,
) -> Self

Pipeline().bucket_auto(by="$price", buckets=5)

def lookup(
    self,
    other: CollectionPipeline | DocumentsPipeline | type[Model],
    /,
    *,
    local_field: FieldLike | None = None,
    foreign_field: FieldLike | None = None,
    into: FieldLike,
) -> Self

# Simple lookup with Model
Pipeline().lookup(Order, local_field="user_id", foreign_field="_id", into="orders")

# Lookup with sub-pipeline
sub = CollectionPipeline("orders").match({"status": "completed"})
Pipeline().lookup(sub, into="completed_orders")

# Lookup with in-memory documents
docs = Pipeline.documents([{"id": 1, "label": "Premium"}])
Pipeline().lookup(docs, local_field="tier_id", foreign_field="id", into="tier")

def graph_lookup(
    self,
    other: type[Model],
    /,
    start_with: FieldLike,
    local_field: FieldLike,
    foreign_field: FieldLike,
    into: FieldLike,
    max_depth: int | None = None,
    depth_field: FieldLike | None = None,
    restrict_search_with_match: MongoQuery | Predicate | None = None,
) -> Self

Pipeline().graph_lookup(
    Employee,
    start_with="$reports_to",
    local_field="reports_to",
    foreign_field="employee_id",
    into="reporting_chain",
    max_depth=5,
)

def facet(self, *facets: Aliased[Pipeline] | dict | list) -> Self

# Dict form
Pipeline().facet({
    "count": Pipeline().count("total"),
    "items": Pipeline().sort({"age": -1}).take(10),
})

# Spread Aliased form
Pipeline().facet(
    Pipeline().count("total").alias("count"),
    Pipeline().sort({"age": -1}).take(10).alias("items"),
)

def union_with(self, other: CollectionPipeline | type[Model], /) -> Self

# Union with a Model class
Pipeline().union_with(ArchivedUser)

# Union with a CollectionPipeline
sub = CollectionPipeline("archived_users").match({"year": 2024})
Pipeline().union_with(sub)

def count(self, output: FieldLike, /) -> Self

Pipeline().match({"status": "active"}).count("active_count")

def sample(self, size: PositiveInteger | None, /) -> Self

Pipeline().sample(5)

def replace_with(self, expr: AnyExpression, /) -> Self

Pipeline().replace_with("$user")
Pipeline().replace_with({"name": "$fullName", "age": "$person_age"})

def set_window_fields(
    self,
    *output: Aliased[WindowOperator] | dict | list,
    sort_by: SortPayload | None = None,
    partition_by: AnyExpression | None = None,
) -> Self

def raw(self, *stages: Stage | Step) -> Self

Pipeline().raw({"$customStage": {"option": "value"}})
Pipeline().raw({"$stage1": {}}, {"$stage2": {}})

def pipe(
    self,
    _0: Callable[[Self, *P.args, **P.kwargs], Self],
    *args: P.args,
    **kwargs: P.kwargs,
) -> Self

def add_active_filter(pipeline: Pipeline) -> Pipeline:
    return pipeline.match(Field("status").eq("active"))

Pipeline().pipe(add_active_filter).sort({"name": 1})

def build(self, *, context: Context | None = None) -> list[Stage]

pipeline = Pipeline().match({"status": "active"}).sort({"age": -1}).take(10)
stages = pipeline.build()
# [{"$match": {"status": "active"}}, {"$sort": {"age": -1}}, {"$limit": 10}]

def alias(self, ref: str) -> Aliased[Self]

Pipeline().count("total").alias("count")
# Used in: Pipeline().facet(Pipeline().count("total").alias("count"))

@classmethod
def documents(cls, *documents: Document | list[Document]) -> DocumentsPipeline

# Spread form
docs = Pipeline.documents(
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
)

# List form
docs = Pipeline.documents([
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
])

# Use with lookup
Pipeline().lookup(docs, local_field="ref_id", foreign_field="id", into="refs")

def add_step(self, step: Step, /) -> Self

@dataclass
class CollectionPipeline(Pipeline)

CollectionPipeline(collection: str, *, steps: list[Step] = [])

sub = CollectionPipeline("orders").match({"status": "completed"}).sort({"date": -1})
Pipeline().lookup(sub, into="recent_orders")

@dataclass
class DocumentsPipeline(Pipeline)

DocumentsPipeline(documents: list[Document], *, steps: list[Step] = [])

docs = Pipeline.documents([{"id": 1}, {"id": 2}])
stages = docs.build()
# [{"$documents": [{"id": 1}, {"id": 2}]}]