47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
from typing import List, Tuple
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
import datetime as dt
|
|
|
|
async def aggregate_salaries(collection: AsyncIOMotorCollection, dt_from: str, dt_upto: str, group_type: str) -> Tuple[List[int], List[str]]: # type: ignore
|
|
group_type_format = "%Y-%m-%dT00:00:00" if group_type == "day" else "%Y-%m-01T00:00:00" if group_type == "month" else "%Y-%m-%dT%H:00:00"
|
|
pipeline = [
|
|
{
|
|
"$match": {
|
|
"dt": {
|
|
"$gt": iso(dt_from),
|
|
"$lt": iso(dt_upto)
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"$group": {
|
|
"_id": {
|
|
"$dateToString":
|
|
{
|
|
"format": group_type_format,
|
|
"date": "$dt"
|
|
}
|
|
},
|
|
"sum": {
|
|
"$sum": "$value"
|
|
},
|
|
}
|
|
},
|
|
{
|
|
"$sort": {
|
|
"_id": 1
|
|
}
|
|
}
|
|
]
|
|
|
|
dataset = []
|
|
labels = []
|
|
|
|
async for document in collection.aggregate(pipeline):
|
|
dataset.append(document["sum"])
|
|
labels.append(document["_id"] )
|
|
|
|
return {"dataset": dataset, "labels": labels}
|
|
|
|
def iso(date: str) -> dt.datetime:
|
|
return dt.datetime.fromisoformat(date) |