Aggregationspipelines og komplekse forespørgsler
Kør filtrerings-, grupperings- og aggregationspipelines effektivt for at understøtte endpoints i analysetil.
Aggregationspipelines og komplekse forespørgsler er en gratis Bootcamp i FastAPI-backendudvikling-lektion på CoddyKit. Dette er lektion 3 af 4. Du kan læse alle 3 lektioner i dette læringsspor gratis i deres fulde længde — derefter låser CoddyKit PRO alle lektioner op samt praktiske øvelser med en indbygget kodeeditor og en AI-underviser døgnet rundt. Den er en del af læringsforløbet i Bootcamp i FastAPI-backendudvikling, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. Bootcamp i FastAPI-backendudvikling-kurset indeholder 4 lektioner i alt.
Hvorfor aggregeringspipelines?
Simple find-forespørgsler returnerer dokumenter uændret. Analyseendepunkter har derimod ofte brug for grupperede, beregnede og omformede data: samlet omsætning pr. måned, gennemsnitlig bedømmelse pr. produkt eller de 10 mest aktive brugere.
MongoDBs aggregeringspipeline udfører dette arbejde i databasen, så du kun sender det endelige resultat over forbindelsen i stedet for at hente tusindvis af dokumenter ind i Python og gennemløbe dem.
- Pipeline = en ordnet liste med trin
- Hvert trin modtager en strøm af dokumenter og sender dokumenter videre
- Beanie gør den tilgængelig via
Document.aggregate(pipeline)
Vores Beanie-modeller
Gennem hele lektionen bruger vi et Order-dokument. Hver ordre har en kunde, en status, et samlet beløb og et oprettelsestidspunkt. Vi bygger analyseendepunkter oven på det.
Beanie-dokumenter nedarver fra beanie.Document, som selv er en Pydantic-model knyttet til 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"$match-trinnet
$match filtrerer dokumenter, præcis som en find-forespørgsel. Placér det så tidligt som muligt, så senere trin behandler færre dokumenter, og eventuelle indeks kan bruges.
Denne pipeline beholder kun ordrer med status paid. Beanies aggregate modtager en almindelig liste med dictionaries og kører den mod samlingen.
pipeline = [
{"$match": {"status": "paid"}}
]
paid_orders = await Order.aggregate(pipeline).to_list()$group-trinnet
$group er kernen i analyser. Det opdeler dokumenter i grupper ud fra et _id-udtryk og beregner akkumulatorer for hver gruppe.
$sum— summen af et felt (eller antal med$sum: 1)$avg,$min,$max$push/$addToSet— saml værdier i et array
Her beregner vi samlet omsætning og antal ordrer pr. kunde.
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}, ...]Feltstier kontra litteraler
Inde i aggregeringsudtryk er en tekststreng, der starter med $, en feltsti (læs værdien af feltet). En almindelig tekststreng er en litteral.
"$total"→ værdien af feltettotal"total"→ den litterale tekststreng "total"{"$sum": 1}→ læg den litterale værdi1til for hvert dokument = et antal
At forveksle dem er begynderes fejl nummer ét. $sum: "$total" summerer beløb; $sum: 1 tæller rækker.
Sortering og begrænsning af resultater
Tilføj $sort og $limit efter grupperingen for at opbygge en rangliste. Sortering bruger 1 for stigende og -1 for faldende rækkefølge.
Dette returnerer de 5 kunder med størst omsætning — en klassisk nyttelast til et analyseendepunkt.
pipeline = [
{"$match": {"status": "paid"}},
{"$group": {
"_id": "$customer_id",
"revenue": {"$sum": "$total"},
}},
{"$sort": {"revenue": -1}},
{"$limit": 5},
]
top_customers = await Order.aggregate(pipeline).to_list()Omformning med $project
$project vælger, hvilke felter der skal bevares, og lader dig omdøbe felter eller beregne nye. Efter et $group ligger gruppenøglen i _id, hvilket sjældent er det navn, dine API-forbrugere forventer.
Her omdøber vi _id til customer_id og fjerner det standardmæssige _id fra resultatet.
pipeline = [
{"$group": {
"_id": "$customer_id",
"revenue": {"$sum": "$total"},
}},
{"$project": {
"_id": 0,
"customer_id": "$_id",
"revenue": 1,
}},
]Tilknytning af resultater til en Pydantic-model
Aggregering returnerer rå dictionaries, ikke Order-dokumenter (strukturen er ændret). Angiv en projection_model, så Beanie validerer hver række til en typet Pydantic-model — perfekt til en FastAPI-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]Gruppering efter dato med $dateToString
Ved tidsserieanalyser kan du gruppere efter en formateret dato. $dateToString omdanner et tidsstempel til en tekststreng, der fungerer som gruppe, for eksempel "2026-06" for månedlig omsætning.
Resultatet er ideelt til endepunkter, der leverer data til diagrammer: én række pr. måned, sorteret kronologisk.
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()Tilslutning til et FastAPI-endepunkt
Placér pipelinen bag en asynkron rute. Fordi aggregeringen kører i MongoDB, forbliver håndteringen lille og hurtig, selv med millioner af ordrer.
Ved at bruge projection_model som response_model får du automatisk validering og 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()Modellering af en pipeline i ren Python
Pipelin-mønstret — match, gruppér, summér — er blot datatransformation. Her er den samme logik i almindelig Python, så du kan se, hvad MongoDB gør internt: filtrere, gruppere efter nøgle og akkumulere en sum.
I produktion gør MongoDB dette langt hurtigere og med indeks, men en forståelse af strukturen hjælper dig med at skrive korrekte trin.
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}")Hurtigt tjek
Du vil have den samlede omsætning pr. kunde, hvor kun betalte ordrer tælles. Hvilken pipeline er korrekt?
Opsummering
Du kan nu opbygge analyseorienterede endepunkter med Beanied aggregeringspipelines:
- $match tidligt for at filtrere og bruge indeks
- $group med akkumulatorer (
$sum,$avg,$sum: 1til optællinger) - Husk, at feltstier skal begynde med
$; almindelige tekststrenge er litteraler - $sort + $limit til ranglister, $project til at omforme og omdøbe
_id - $dateToString til tidsseriegrupper
- Angiv
projection_modelfor at tilknytte rå rækker til typede Pydantic-modeller og få et rent FastAPI-response_model-resultat
Pipелinen kører i MongoDB, så dine håndteringer forbliver små, og dine endepunkter hurtige.
Lær Bootcamp i FastAPI-backendudvikling med en AI-underviser — gratis
Skriv og kør rigtig kode i din browser, få øjeblikkelig hjælp fra en AI-underviser døgnet rundt, og fortsæt, hvor du slap, på web eller i appen.
- Kurser
- 21
- Lektioner
- 84
Ofte stillede spørgsmål
Er lektionen “Aggregationspipelines og komplekse forespørgsler” gratis?
Ja — alle 3 lektioner i læringssporet Bootcamp i FastAPI-backendudvikling, inklusive “Aggregationspipelines og komplekse forespørgsler”, kan læses gratis i deres fulde længde her på webstedet. Derefter låser CoddyKit PRO alle lektioner op samt interaktive øvelser med en indbygget kodeeditor og en AI-underviser døgnet rundt. Bootcamp i FastAPI-backendudvikling-kurset indeholder 4 lektioner i alt.
Hvad lærer jeg i “Aggregationspipelines og komplekse forespørgsler”?
Kør filtrerings-, grupperings- og aggregationspipelines effektivt for at understøtte endpoints i analysetil. Du øver dig i Bootcamp i FastAPI-backendudvikling med praktisk kode, som du kører direkte i browseren, og en AI-vejleder døgnet rundt besvarer dine spørgsmål, mens du arbejder dig gennem lektionen.
Skal jeg have erfaring for at begynde på Bootcamp i FastAPI-backendudvikling?
Der kræves ingen tidligere erfaring. Bootcamp i FastAPI-backendudvikling på CoddyKit er tilrettelagt for både begyndere og øvede, så du kan starte her eller fra begyndelsen og lære i dit eget tempo. Dette er lektion 3 af 4.
Hvor lang tid tager lektionen “Aggregationspipelines og komplekse forespørgsler”?
De fleste CoddyKit-lektioner tager cirka 5–10 minutter. Hver lektion er kort og interaktiv, så du gør løbende fremskridt og kan fortsætte, hvor du slap – på både web og app.
Kan jeg skrive og køre kode i denne Bootcamp i FastAPI-backendudvikling-lektion?
Ja. Alle Bootcamp i FastAPI-backendudvikling-lektioner har en indbygget kodeeditor, så du kan skrive og køre rigtig kode direkte i din browser og få øjeblikkelig feedback fra AI – uden lokal opsætning.
Alle lektioner i dette kursus
- Asynkron MongoDB-adgang med Motor
- Dokumentmodellering med Beanie ODM
- Aggregationspipelines og komplekse forespørgsler
- Skemaudvikling og dokumentmigreringer