feat: information pages — CRUD model, admin API, public API

- InfoPage model: slug, title (JSONB locale dict), content (JSONB),
  is_active, sort_order, icon, created_at/updated_at
- CRUD: create, get by id/slug, list, update, delete, reorder
- Admin routes: /admin/info-pages with full CRUD, toggle-active, reorder
  (permissions: settings:read/settings:edit)
- Public routes: /info-pages list active, /info-pages/{slug} get by slug
- Migration 0065: create info_pages table with unique slug index
- Custom pages support: admins can create any info page with any slug
This commit is contained in:
Fringg
2026-04-24 08:09:54 +03:00
parent 59c54c9b39
commit e4b4a54797
7 changed files with 550 additions and 0 deletions
+162
View File
@@ -0,0 +1,162 @@
"""CRUD operations for info pages."""
from datetime import UTC, datetime
from typing import Any
import structlog
from sqlalchemy import delete, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.models import InfoPage
logger = structlog.get_logger(__name__)
# Fields that can be set via update_info_page
_ALLOWED_UPDATE_FIELDS: frozenset[str] = frozenset(
{
'slug',
'title',
'content',
'is_active',
'sort_order',
'icon',
}
)
# Fields that can be explicitly set to None
_NULLABLE_UPDATE_FIELDS: frozenset[str] = frozenset(
{
'icon',
}
)
async def create_info_page(
db: AsyncSession,
*,
slug: str,
title: dict[str, str],
content: dict[str, str],
is_active: bool = True,
sort_order: int = 0,
icon: str | None = None,
) -> InfoPage:
"""Create a new info page.
Raises:
IntegrityError: if slug is not unique (caller must handle).
"""
page = InfoPage(
slug=slug,
title=title,
content=content,
is_active=is_active,
sort_order=sort_order,
icon=icon,
)
db.add(page)
try:
await db.commit()
except IntegrityError:
await db.rollback()
raise
await db.refresh(page)
logger.info('Created info page', page_id=page.id, slug=page.slug)
return page
async def get_info_page_by_id(db: AsyncSession, page_id: int) -> InfoPage | None:
"""Get an info page by ID."""
result = await db.execute(select(InfoPage).where(InfoPage.id == page_id))
return result.scalar_one_or_none()
async def get_info_page_by_slug(db: AsyncSession, slug: str) -> InfoPage | None:
"""Get an info page by slug."""
result = await db.execute(select(InfoPage).where(InfoPage.slug == slug))
return result.scalar_one_or_none()
async def get_all_info_pages(
db: AsyncSession,
*,
include_inactive: bool = False,
) -> list[InfoPage]:
"""Get all info pages, ordered by sort_order ascending."""
stmt = select(InfoPage)
if not include_inactive:
stmt = stmt.where(InfoPage.is_active.is_(True))
stmt = stmt.order_by(InfoPage.sort_order.asc(), InfoPage.id.asc())
result = await db.execute(stmt)
return list(result.scalars().all())
async def update_info_page(
db: AsyncSession,
page_id: int,
**kwargs: Any,
) -> InfoPage | None:
"""Update an info page. Only whitelisted fields are applied.
Raises:
IntegrityError: if slug conflicts with another page (caller must handle).
"""
update_data: dict[str, Any] = {}
for key, value in kwargs.items():
if key not in _ALLOWED_UPDATE_FIELDS:
continue
if value is None and key not in _NULLABLE_UPDATE_FIELDS:
continue
update_data[key] = value
if not update_data:
return await get_info_page_by_id(db, page_id)
update_data['updated_at'] = datetime.now(UTC)
await db.execute(update(InfoPage).where(InfoPage.id == page_id).values(**update_data))
try:
await db.commit()
except IntegrityError:
await db.rollback()
raise
page = await get_info_page_by_id(db, page_id)
if page:
logger.info(
'Updated info page',
page_id=page_id,
updated_fields=list(update_data.keys()),
)
return page
async def delete_info_page(db: AsyncSession, page_id: int) -> None:
"""Delete an info page."""
await db.execute(delete(InfoPage).where(InfoPage.id == page_id))
await db.commit()
logger.info('Deleted info page', page_id=page_id)
async def reorder_info_pages(db: AsyncSession, items: list[dict]) -> None:
"""Bulk update sort_order for info pages.
Each dict in *items* must have ``id`` and ``sort_order`` keys.
"""
for item in items:
page_id = item.get('id')
sort_order = item.get('sort_order')
if page_id is None or sort_order is None:
continue
await db.execute(
update(InfoPage).where(InfoPage.id == page_id).values(sort_order=sort_order, updated_at=datetime.now(UTC))
)
await db.commit()
logger.info('Reordered info pages', count=len(items))
+17
View File
@@ -3699,3 +3699,20 @@ class YandexClientIdMap(Base):
subid = Column(String(255), nullable=True)
created_at = Column(AwareDateTime(), server_default=func.now())
updated_at = Column(AwareDateTime(), server_default=func.now(), onupdate=func.now())
class InfoPage(Base):
"""Static informational page with multilingual title/content (JSONB)."""
__tablename__ = 'info_pages'
__table_args__ = (Index('ix_info_pages_slug', 'slug', unique=True),)
id = Column(Integer, primary_key=True, index=True)
slug = Column(String(200), unique=True, nullable=False, index=True)
title = Column(JSONB, nullable=False, server_default='{}')
content = Column(JSONB, nullable=False, server_default='{}')
is_active = Column(Boolean, nullable=False, default=True, server_default='true')
sort_order = Column(Integer, nullable=False, default=0, server_default='0')
icon = Column(String(50), nullable=True)
created_at = Column(AwareDateTime(), server_default=func.now())
updated_at = Column(AwareDateTime(), server_default=func.now(), onupdate=func.now())