Bootcamp i backendutveckling med FastAPI · Lektion

Aggregeringspipelines och komplexa queries

Kör filtrerings-, grupperings- och aggregeringspipelines för att effektivt driva analysliknande endpoints.

Lektion 3 av 413 steg

Aggregeringspipelines och komplexa queries är en gratis lektion i Bootcamp i backendutveckling med FastAPI på CoddyKit. Detta är lektion 3 av 4. Du kan läsa vilka 3 lektioner som helst i den här lärvägen kostnadsfritt i sin helhet – därefter låser CoddyKit PRO upp alla lektioner, plus praktisk övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Den ingår i lärvägen för Bootcamp i backendutveckling med FastAPI, och Era framsteg synkroniseras mellan webben och CoddyKit-appen. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Varför använda aggregeringspipelines?

Enkla find-frågor returnerar dokument som de är. Analysendpoints behöver däremot ofta grupperade, beräknade och omformade data: totala intäkter per månad, genomsnittligt betyg per produkt eller de tio mest aktiva användarna.

MongoDB:s aggregeringspipeline utför detta arbete i databasen, så du skickar bara slutresultatet över nätverket i stället för att hämta tusentals dokument till Python och iterera över dem.

Pipeline = en ordnad lista med steg
Varje steg tar emot en ström av dokument och skickar ut dokument
Beanie exponerar detta via Document.aggregate(pipeline)

Våra Beanie-modeller

Genom hela lektionen använder vi ett Order-dokument. Varje order har en kund, en status, ett totalbelopp och en tidpunkt för skapandet. Vi bygger analysendpoints ovanpå det.

Beanie-dokument ärver från beanie.Document, som i sin tur är en Pydantic-modell kopplad till en MongoDB-samling.

from datetime import datetime
from beanie import Document
from pydantic import Field

class Order(Document):
    customer_id: str
    status: str  # "paid", "pending", "cancelled"
    total: float
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "orders"

Steget $match

$match filtrerar dokument, precis som en find-fråga. Placera det så tidigt som möjligt så att senare steg behandlar färre dokument och eventuella index kan användas.

Den här pipelinen behåller endast paid-ordrar. Beaniels aggregate tar emot en vanlig lista med dict-objekt och kör den mot samlingen.

pipeline = [
    {"$match": {"status": "paid"}}
]

paid_orders = await Order.aggregate(pipeline).to_list()

Steget $group

$group är kärnan i analysarbete. Det delar in dokument i grupper utifrån ett _id-uttryck och beräknar ackumulatorer för varje grupp.

$sum – summan av ett fält (eller antal med $sum: 1)
$avg, $min, $max
$push / $addToSet – samlar värden i en array

Här beräknar vi totala intäkter och antal ordrar per kund.

pipeline = [
    {"$match": {"status": "paid"}},
    {"$group": {
        "_id": "$customer_id",
        "revenue": {"$sum": "$total"},
        "order_count": {"$sum": 1},
    }},
]

rows = await Order.aggregate(pipeline).to_list()
# [{"_id": "c1", "revenue": 240.0, "order_count": 3}, ...]

Fältsökvägar jämfört med literaler

I aggregeringsuttryck är en sträng som börjar med $ en fältsökväg (läs värdet i fältet). En vanlig sträng är en literal.

"$total" → värdet i fältet total
"total" → den literala strängen "total"
{"$sum": 1} → lägg till den literala 1 för varje dokument = en räkning

Att blanda ihop dessa är nybörjarmisstag nummer ett. $sum: "$total" summerar belopp; $sum: 1 räknar rader.

Sortera och begränsa resultat

Lägg till $sort och $limit efter grupperingen för att skapa en topplista. Sorteringen använder 1 för stigande och -1 för fallande ordning.

Detta returnerar de fem kunder som har högst intäkter – ett klassiskt resultat för en analysendpoint.

pipeline = [
    {"$match": {"status": "paid"}},
    {"$group": {
        "_id": "$customer_id",
        "revenue": {"$sum": "$total"},
    }},
    {"$sort": {"revenue": -1}},
    {"$limit": 5},
]

top_customers = await Order.aggregate(pipeline).to_list()

Forma om med $project

$project väljer vilka fält som ska behållas och låter dig byta namn på eller beräkna nya fält. Efter ett $group finns gruppnyckeln i _id, vilket sällan är det namn som API-konsumenterna förväntar sig.

Här byter vi namn på _id till customer_id och tar bort standardfältet _id från resultatet.

pipeline = [
    {"$group": {
        "_id": "$customer_id",
        "revenue": {"$sum": "$total"},
    }},
    {"$project": {
        "_id": 0,
        "customer_id": "$_id",
        "revenue": 1,
    }},
]

Mappa resultat till en Pydantic-modell

Aggregering returnerar råa dict-objekt, inte Order-dokument (strukturen har ändrats). Ange en projection_model så att Beanie validerar varje rad mot en typad Pydantic-modell – perfekt för FastAPIs response_model.

from pydantic import BaseModel

class CustomerRevenue(BaseModel):
    customer_id: str
    revenue: float

pipeline = [
    {"$group": {"_id": "$customer_id", "revenue": {"$sum": "$total"}}},
    {"$project": {"_id": 0, "customer_id": "$_id", "revenue": 1}},
    {"$sort": {"revenue": -1}},
]

results = await Order.aggregate(
    pipeline, projection_model=CustomerRevenue
).to_list()  # List[CustomerRevenue]

Gruppera efter datum med $dateToString

För tidsserieanalyser grupperar du efter ett formaterat datum. $dateToString omvandlar en tidsstämpel till en strängnyckel, till exempel "2026-06" för månadsintäkter.

Resultatet passar utmärkt för endpoints som ritar diagram: en rad per månad, sorterad kronologiskt.

pipeline = [
    {"$match": {"status": "paid"}},
    {"$group": {
        "_id": {"$dateToString": {
            "format": "%Y-%m", "date": "$created_at"
        }},
        "revenue": {"$sum": "$total"},
    }},
    {"$sort": {"_id": 1}},
]

monthly = await Order.aggregate(pipeline).to_list()

Koppla in det i en FastAPI-endpoint

Lägg pipelinen bakom en asynkron route. Eftersom aggregeringen körs i MongoDB förblir hanteraren liten och snabb även med miljontals ordrar.

Om du använder projection_model som response_model får du automatisk validering och OpenAPI-dokumentation.

from fastapi import APIRouter

router = APIRouter()

@router.get("/analytics/top-customers", response_model=list[CustomerRevenue])
async def top_customers(limit: int = 5):
    pipeline = [
        {"$match": {"status": "paid"}},
        {"$group": {"_id": "$customer_id", "revenue": {"$sum": "$total"}}},
        {"$project": {"_id": 0, "customer_id": "$_id", "revenue": 1}},
        {"$sort": {"revenue": -1}},
        {"$limit": limit},
    ]
    return await Order.aggregate(pipeline, projection_model=CustomerRevenue).to_list()

Modellera en pipeline i ren Python

Pipelinemönstret – matcha, gruppera, summera – är bara datatransformering. Här är samma logik i vanlig Python, så att du kan se vad MongoDB gör internt: filtrera, gruppera efter nyckel och ackumulera en summa.

I produktion gör MongoDB detta mycket snabbare och med hjälp av index, men om du förstår strukturen blir det lättare att skriva korrekta steg.

orders = [
    {"customer_id": "c1", "status": "paid", "total": 100.0},
    {"customer_id": "c2", "status": "paid", "total": 40.0},
    {"customer_id": "c1", "status": "paid", "total": 60.0},
    {"customer_id": "c2", "status": "pending", "total": 999.0},
]

revenue = {}
for o in orders:
    if o["status"] != "paid":  # $match
        continue
    revenue[o["customer_id"]] = revenue.get(o["customer_id"], 0) + o["total"]  # $group + $sum

top = sorted(revenue.items(), key=lambda kv: kv[1], reverse=True)  # $sort
for customer_id, total in top:
    print(f"{customer_id}: {total}")

Snabbkontroll

Du vill ha totala intäkter per kund och endast räkna betalda ordrar. Vilken pipeline är korrekt?

Besvara frågan i den interaktiva lektionen

Sammanfattning

Nu kan du bygga analysliknande endpoints med Beaniels aggregeringspipelines:

$match tidigt för att filtrera och använda index
$group med ackumulatorer ($sum, $avg, $sum: 1 för antal)
Kom ihåg att fältsökvägar behöver ett inledande $; vanliga strängar är literaler
$sort + $limit för topplistor, $project för att forma om och byta namn på _id
$dateToString för tidsserienycklar
Ange projection_model för att mappa råa rader till typade Pydantic-modeller och få ett tydligt FastAPI-response_model-resultat

Pipelinen körs i MongoDB, vilket håller dina hanterare små och dina endpoints snabba.

Gratis att börja

Lär dig Bootcamp i backendutveckling med FastAPI med en AI-lärare – gratis

Skriv och kör riktig kod i webbläsaren, få omedelbar hjälp av en AI-lärare dygnet runt och fortsätt där du slutade – på webben eller i appen.

Kurser: 21
Lektioner: 84

Börja lära dig gratis Visa hela kursen

Vanliga frågor

Är lektionen ”Aggregeringspipelines och komplexa queries” gratis?

Ja – du kan läsa vilka 3 lektioner som helst i lärvägen Bootcamp i backendutveckling med FastAPI, inklusive ”Aggregeringspipelines och komplexa queries”, kostnadsfritt i sin helhet här på webben. Därefter låser CoddyKit PRO upp alla lektioner, plus interaktiv övning med en inbyggd kodredigerare och en AI-lärare dygnet runt. Kursen i Bootcamp i backendutveckling med FastAPI innehåller totalt 4 lektioner.

Vad lär jag mig i ”Aggregeringspipelines och komplexa queries”?

Kör filtrerings-, grupperings- och aggregeringspipelines för att effektivt driva analysliknande endpoints. Ni övar på Bootcamp i backendutveckling med FastAPI med praktisk kod som körs direkt i webbläsaren, medan en AI-handledare som är tillgänglig dygnet runt svarar på Era frågor under lektionen.

Behöver jag någon erfarenhet för att börja lära mig Bootcamp i backendutveckling med FastAPI?

Du behöver inga förkunskaper. Utbildningen i Bootcamp i backendutveckling med FastAPI på CoddyKit är upplagd för allt från nybörjare till avancerade elever, så att du kan börja här eller från början och gå fram i din egen takt. Detta är lektion 3 av 4.

Hur lång tid tar lektionen ”Aggregeringspipelines och komplexa queries”?

De flesta CoddyKit-lektioner tar cirka 5–10 minuter. Varje lektion är kort och interaktiv, så att du gör stadiga framsteg och kan fortsätta precis där du slutade – på webben eller i appen.

Kan jag skriva och köra kod i den här Bootcamp i backendutveckling med FastAPI-lektionen?

Ja. Varje Bootcamp i backendutveckling med FastAPI-lektion innehåller en inbyggd kodredigerare, så att du kan skriva och köra riktig kod direkt i webbläsaren och få omedelbar AI-feedback – utan lokal installation.

Alla lektioner i den här kursen

Asynkron MongoDB-åtkomst med Motor
Dokumentmodellering med Beanie ODM
Aggregeringspipelines och komplexa queries
Utveckling av scheman och dokumentmigreringar

← Tillbaka till Bootcamp i backendutveckling med FastAPI