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>
140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""Publishing / SEO / PWA repositories — Phase 11.7."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Sequence
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.models.publishing import (
|
|
DomainEdgeBinding,
|
|
PublishRelease,
|
|
PwaManifest,
|
|
RobotsShell,
|
|
SeoMetadata,
|
|
SitemapShell,
|
|
)
|
|
from app.models.types import SeoTarget
|
|
from app.repositories.base import TenantBaseRepository
|
|
|
|
|
|
async def _list_filtered(session, model, tenant_id, offset, limit, spec):
|
|
base = select(model).where(
|
|
model.tenant_id == tenant_id,
|
|
model.is_deleted.is_(False),
|
|
)
|
|
count_base = select(func.count()).select_from(model).where(
|
|
model.tenant_id == tenant_id,
|
|
model.is_deleted.is_(False),
|
|
)
|
|
if spec is not None:
|
|
base = spec.apply(base)
|
|
for clause in spec.filter_clauses():
|
|
count_base = count_base.where(clause)
|
|
else:
|
|
base = base.order_by(model.created_at.desc())
|
|
total = int((await session.execute(count_base)).scalar_one())
|
|
result = await session.execute(base.offset(offset).limit(limit))
|
|
return result.scalars().all(), total
|
|
|
|
|
|
class PublishReleaseRepository(TenantBaseRepository[PublishRelease]):
|
|
model = PublishRelease
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|
|
|
|
|
|
class DomainEdgeBindingRepository(TenantBaseRepository[DomainEdgeBinding]):
|
|
model = DomainEdgeBinding
|
|
|
|
async def get_by_site_domain(
|
|
self, tenant_id: UUID, site_domain_id: UUID
|
|
) -> DomainEdgeBinding | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_domain_id == site_domain_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
return (await self.session.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|
|
|
|
|
|
class SeoMetadataRepository(TenantBaseRepository[SeoMetadata]):
|
|
model = SeoMetadata
|
|
|
|
async def get_by_target(
|
|
self, tenant_id: UUID, target_type: SeoTarget, target_id: UUID
|
|
) -> SeoMetadata | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.target_type == target_type,
|
|
self.model.target_id == target_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
return (await self.session.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|
|
|
|
|
|
class SitemapShellRepository(TenantBaseRepository[SitemapShell]):
|
|
model = SitemapShell
|
|
|
|
async def get_by_site(
|
|
self, tenant_id: UUID, site_id: UUID
|
|
) -> SitemapShell | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
return (await self.session.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|
|
|
|
|
|
class RobotsShellRepository(TenantBaseRepository[RobotsShell]):
|
|
model = RobotsShell
|
|
|
|
async def get_by_site(self, tenant_id: UUID, site_id: UUID) -> RobotsShell | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
return (await self.session.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|
|
|
|
|
|
class PwaManifestRepository(TenantBaseRepository[PwaManifest]):
|
|
model = PwaManifest
|
|
|
|
async def get_by_site(self, tenant_id: UUID, site_id: UUID) -> PwaManifest | None:
|
|
stmt = select(self.model).where(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.site_id == site_id,
|
|
self.model.is_deleted.is_(False),
|
|
)
|
|
return (await self.session.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def list_filtered(self, tenant_id, *, offset, limit, spec=None):
|
|
return await _list_filtered(
|
|
self.session, self.model, tenant_id, offset, limit, spec
|
|
)
|