5c6a2e78cf
Register all active platform services and base features in core DB so admin can manage them, add delivery/hospitality/sports-center backend phases, update apps catalog and production deploy tooling. Co-authored-by: Cursor <cursoragent@cursor.com>
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""Analytics repositories — Phase 12.8."""
|
|
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.models.analytics import AnalyticsReportDefinition, AnalyticsSnapshot
|
|
from app.repositories.base import TenantBaseRepository
|
|
|
|
|
|
class AnalyticsReportDefinitionRepository(TenantBaseRepository[AnalyticsReportDefinition]):
|
|
model = AnalyticsReportDefinition
|
|
|
|
async def get_by_code(
|
|
self, tenant_id: UUID, code: str
|
|
) -> AnalyticsReportDefinition | None:
|
|
stmt = select(AnalyticsReportDefinition).where(
|
|
AnalyticsReportDefinition.tenant_id == tenant_id,
|
|
AnalyticsReportDefinition.code == code,
|
|
AnalyticsReportDefinition.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
class AnalyticsSnapshotRepository(TenantBaseRepository[AnalyticsSnapshot]):
|
|
model = AnalyticsSnapshot
|
|
|
|
async def list_for_report(
|
|
self, tenant_id: UUID, report_id: UUID, *, offset: int, limit: int
|
|
) -> list[AnalyticsSnapshot]:
|
|
stmt = (
|
|
select(AnalyticsSnapshot)
|
|
.where(
|
|
AnalyticsSnapshot.tenant_id == tenant_id,
|
|
AnalyticsSnapshot.report_id == report_id,
|
|
AnalyticsSnapshot.is_deleted.is_(False),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|