106 lines
2.8 KiB
Python
106 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.database import session_scope
|
|
from app.modules.content.models import ContentPage
|
|
|
|
ALLOWED_PAGE_STATUSES = {"draft", "published"}
|
|
|
|
|
|
def _detach(db, instance):
|
|
db.refresh(instance)
|
|
db.expunge(instance)
|
|
return instance
|
|
|
|
|
|
def list_published_pages() -> list[ContentPage]:
|
|
with session_scope() as db:
|
|
pages = list(
|
|
db.scalars(select(ContentPage).where(ContentPage.status == "published").order_by(ContentPage.slug))
|
|
)
|
|
return [_detach(db, page) for page in pages]
|
|
|
|
|
|
def list_all_pages() -> list[ContentPage]:
|
|
with session_scope() as db:
|
|
pages = list(db.scalars(select(ContentPage).order_by(ContentPage.slug)))
|
|
return [_detach(db, page) for page in pages]
|
|
|
|
|
|
def get_page_by_slug(slug: str, include_draft: bool = False) -> ContentPage | None:
|
|
with session_scope() as db:
|
|
page = db.scalar(select(ContentPage).where(ContentPage.slug == slug))
|
|
if not page:
|
|
return None
|
|
if include_draft or page.status == "published":
|
|
return _detach(db, page)
|
|
return None
|
|
|
|
|
|
def get_page_by_id(page_id: str) -> ContentPage | None:
|
|
with session_scope() as db:
|
|
page = db.get(ContentPage, page_id)
|
|
if not page:
|
|
return None
|
|
return _detach(db, page)
|
|
|
|
|
|
def create_page(
|
|
slug: str,
|
|
title: str,
|
|
body: str,
|
|
status: str,
|
|
author_id: str,
|
|
) -> ContentPage:
|
|
if status not in ALLOWED_PAGE_STATUSES:
|
|
raise ValueError("INVALID_STATUS")
|
|
with session_scope() as db:
|
|
page = ContentPage(
|
|
id=str(uuid4()),
|
|
slug=slug,
|
|
title=title,
|
|
body=body,
|
|
status=status,
|
|
author_id=author_id,
|
|
published_at=datetime.now(UTC) if status == "published" else None,
|
|
)
|
|
db.add(page)
|
|
db.flush()
|
|
return _detach(db, page)
|
|
|
|
|
|
def update_page(
|
|
page_id: str,
|
|
title: str | None,
|
|
body: str | None,
|
|
status: str | None,
|
|
) -> ContentPage:
|
|
with session_scope() as db:
|
|
page = db.get(ContentPage, page_id)
|
|
if not page:
|
|
raise KeyError(page_id)
|
|
if title:
|
|
page.title = title
|
|
if body:
|
|
page.body = body
|
|
if status:
|
|
if status not in ALLOWED_PAGE_STATUSES:
|
|
raise ValueError("INVALID_STATUS")
|
|
page.status = status
|
|
if status == "published" and page.published_at is None:
|
|
page.published_at = datetime.now(UTC)
|
|
page.updated_at = datetime.now(UTC)
|
|
db.flush()
|
|
return _detach(db, page)
|
|
|
|
|
|
def delete_page(page_id: str) -> None:
|
|
with session_scope() as db:
|
|
page = db.get(ContentPage, page_id)
|
|
if page:
|
|
db.delete(page)
|