203671a7bf
Add the experience service with sites through analytics/AI hooks, migrations through 0011, and phase docs/manifests marking the track complete. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Sites & page repositories — Phase 11.1."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Sequence
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.models.sites import ExperiencePage, ExperienceSite, ExperienceSiteDomain
|
|
from app.repositories.base import TenantBaseRepository
|
|
|
|
|
|
class ExperienceSiteRepository(TenantBaseRepository[ExperienceSite]):
|
|
model = ExperienceSite
|
|
|
|
async def get_by_code(
|
|
self, tenant_id: UUID, workspace_id: UUID, code: str
|
|
) -> ExperienceSite | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.workspace_id == workspace_id,
|
|
self.model.code == code,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_workspace(
|
|
self, tenant_id: UUID, workspace_id: UUID, *, offset: int = 0, limit: int = 20
|
|
) -> Sequence[ExperienceSite]:
|
|
stmt = (
|
|
select(self.model)
|
|
.where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.workspace_id == workspace_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
class ExperiencePageRepository(TenantBaseRepository[ExperiencePage]):
|
|
model = ExperiencePage
|
|
|
|
async def get_by_code(
|
|
self, tenant_id: UUID, site_id: UUID, code: str
|
|
) -> ExperiencePage | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.code == code,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_slug(
|
|
self, tenant_id: UUID, site_id: UUID, slug: str
|
|
) -> ExperiencePage | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.slug == slug,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_site(
|
|
self, tenant_id: UUID, site_id: UUID, *, offset: int = 0, limit: int = 20
|
|
) -> Sequence[ExperiencePage]:
|
|
stmt = (
|
|
select(self.model)
|
|
.where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
class ExperienceSiteDomainRepository(TenantBaseRepository[ExperienceSiteDomain]):
|
|
model = ExperienceSiteDomain
|
|
|
|
async def get_by_host(
|
|
self, tenant_id: UUID, host: str
|
|
) -> ExperienceSiteDomain | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.host == host,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_site(
|
|
self, tenant_id: UUID, site_id: UUID, *, offset: int = 0, limit: int = 20
|
|
) -> Sequence[ExperienceSiteDomain]:
|
|
stmt = (
|
|
select(self.model)
|
|
.where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|