45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from app.models.mongodb.analytics_result import AnalyticsResult
|
|
|
|
|
|
class AnalyticsRepository:
|
|
async def create(self, result: AnalyticsResult) -> AnalyticsResult:
|
|
return await result.insert()
|
|
|
|
async def get_by_type(
|
|
self,
|
|
analysis_type: str,
|
|
device_id: str | None = None,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
) -> list[AnalyticsResult]:
|
|
query: dict = {"analysis_type": analysis_type}
|
|
if device_id:
|
|
query["device_id"] = device_id
|
|
return await (
|
|
AnalyticsResult.find(query)
|
|
.sort("-created_at")
|
|
.skip(skip)
|
|
.limit(limit)
|
|
.to_list()
|
|
)
|
|
|
|
async def get_by_period(
|
|
self,
|
|
analysis_type: str,
|
|
start: datetime,
|
|
end: datetime,
|
|
device_id: str | None = None,
|
|
) -> list[AnalyticsResult]:
|
|
query: dict = {
|
|
"analysis_type": analysis_type,
|
|
"period_start": {"$gte": start},
|
|
"period_end": {"$lte": end},
|
|
}
|
|
if device_id:
|
|
query["device_id"] = device_id
|
|
return await AnalyticsResult.find(query).sort("-created_at").to_list()
|