ponzischeme89
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""
|
|
Service layer for Page CRUD operations.
|
|
All DB queries are async; HTML body is sanitized on write.
|
|
"""
|
|
import nh3
|
|
from typing import Optional
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.page import Page
|
|
from app.schemas.page import PageCreate, PageUpdate
|
|
|
|
|
|
def _sanitize_body(body: str) -> str:
|
|
"""Strip dangerous HTML tags/attributes using nh3."""
|
|
return nh3.clean(body)
|
|
|
|
|
|
async def get_published_pages(db: AsyncSession) -> list[Page]:
|
|
result = await db.execute(
|
|
select(Page).where(Page.published == True).order_by(Page.created_at.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_page_by_slug(db: AsyncSession, slug: str, published_only: bool = True) -> Optional[Page]:
|
|
stmt = select(Page).where(Page.slug == slug)
|
|
if published_only:
|
|
stmt = stmt.where(Page.published == True)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().first()
|
|
|
|
|
|
async def create_page(db: AsyncSession, data: PageCreate) -> Page:
|
|
page = Page(
|
|
title=data.title,
|
|
slug=data.slug,
|
|
body=_sanitize_body(data.body),
|
|
meta_title=data.meta_title,
|
|
meta_description=data.meta_description,
|
|
og_image_url=data.og_image_url,
|
|
published=data.published,
|
|
)
|
|
db.add(page)
|
|
await db.flush()
|
|
await db.refresh(page)
|
|
return page
|
|
|
|
|
|
async def update_page(db: AsyncSession, slug: str, data: PageUpdate) -> Optional[Page]:
|
|
page = await get_page_by_slug(db, slug, published_only=False)
|
|
if page is None:
|
|
return None
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
if "body" in update_data and update_data["body"] is not None:
|
|
update_data["body"] = _sanitize_body(update_data["body"])
|
|
|
|
for field, value in update_data.items():
|
|
setattr(page, field, value)
|
|
|
|
await db.flush()
|
|
await db.refresh(page)
|
|
return page
|
|
|
|
|
|
async def delete_page(db: AsyncSession, slug: str) -> bool:
|
|
page = await get_page_by_slug(db, slug, published_only=False)
|
|
if page is None:
|
|
return False
|
|
await db.delete(page)
|
|
await db.flush()
|
|
return True
|