sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add CMS service layer and API endpoints

Browse Source 
Implement complete CMS business logic and REST API:

Service Layer (content_page_service.py):
- get_page_for_vendor() - Two-tier lookup with fallback
- list_pages_for_vendor() - Merge vendor + platform pages
- create_page(), update_page(), delete_page() - CRUD operations
- Support for published/draft workflow
- Footer/header navigation filtering

API Endpoints:

Admin API (/api/v1/admin/content-pages):
- POST /platform - Create platform defaults
- GET /platform - List platform defaults
- GET / - List all pages with vendor filtering
- PUT /{id} - Update any page
- DELETE /{id} - Delete any page

Vendor API (/api/v1/vendor/{code}/content-pages):
- GET / - List available pages (vendor + platform merged)
- GET /overrides - List only vendor overrides
- POST / - Create vendor override
- PUT /{id} - Update vendor page
- DELETE /{id} - Delete vendor page

Shop API (/api/v1/shop/content-pages):
- GET /navigation - Get footer/header navigation pages
- GET /{slug} - Get specific page (public, with fallback)

All endpoints include proper authentication, authorization,
and validation using Pydantic schemas.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-11-22 15:54:48 +01:00
parent c219f5b5f8
commit fb3aa89086
8 changed files with 1043 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/api/main.py
View File

@@ -9,7 +9,7 @@ This module provides:
"""
from fastapi import APIRouter
from app.api.v1 import admin, vendor, public
from app.api.v1 import admin, vendor, public, shop
api_router = APIRouter()
@@ -46,3 +46,14 @@ api_router.include_router(
tags=["public"]
)
# ============================================================================
# SHOP ROUTES (Public shop frontend API)
# Prefix: /api/v1/shop
# ============================================================================
api_router.include_router(
shop.router,
prefix="/v1/shop",
tags=["shop"]
)

6
app/api/v1/admin/__init__.py
View File

@@ -34,7 +34,8 @@ from . import (
monitoring,
audit,
settings,
notifications
notifications,
content_pages
)
# Create admin router
@@ -62,6 +63,9 @@ router.include_router(vendor_domains.router, tags=["admin-vendor-domains"])
# Include vendor themes management endpoints
router.include_router(vendor_themes.router, tags=["admin-vendor-themes"])
# Include content pages management endpoints
router.include_router(content_pages.router, prefix="/content-pages", tags=["admin-content-pages"])
# ============================================================================
# User Management

232
app/api/v1/admin/content_pages.py Normal file
View File

@@ -0,0 +1,232 @@
# app/api/v1/admin/content_pages.py
"""
Admin Content Pages API
Platform administrators can:
- Create/edit/delete platform default content pages
- View all vendor content pages
- Override vendor content if needed
"""
import logging
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user, get_db
from app.services.content_page_service import content_page_service
from models.database.user import User
router = APIRouter()
logger = logging.getLogger(__name__)
# ============================================================================
# REQUEST/RESPONSE SCHEMAS
# ============================================================================
class ContentPageCreate(BaseModel):
"""Schema for creating a content page."""
slug: str = Field(..., max_length=100, description="URL-safe identifier (about, faq, contact, etc.)")
title: str = Field(..., max_length=200, description="Page title")
content: str = Field(..., description="HTML or Markdown content")
content_format: str = Field(default="html", description="Content format: html or markdown")
meta_description: Optional[str] = Field(None, max_length=300, description="SEO meta description")
meta_keywords: Optional[str] = Field(None, max_length=300, description="SEO keywords")
is_published: bool = Field(default=False, description="Publish immediately")
show_in_footer: bool = Field(default=True, description="Show in footer navigation")
show_in_header: bool = Field(default=False, description="Show in header navigation")
display_order: int = Field(default=0, description="Display order (lower = first)")
vendor_id: Optional[int] = Field(None, description="Vendor ID (None for platform default)")
class ContentPageUpdate(BaseModel):
"""Schema for updating a content page."""
title: Optional[str] = Field(None, max_length=200)
content: Optional[str] = None
content_format: Optional[str] = None
meta_description: Optional[str] = Field(None, max_length=300)
meta_keywords: Optional[str] = Field(None, max_length=300)
is_published: Optional[bool] = None
show_in_footer: Optional[bool] = None
show_in_header: Optional[bool] = None
display_order: Optional[int] = None
class ContentPageResponse(BaseModel):
"""Schema for content page response."""
id: int
vendor_id: Optional[int]
vendor_name: Optional[str]
slug: str
title: str
content: str
content_format: str
meta_description: Optional[str]
meta_keywords: Optional[str]
is_published: bool
published_at: Optional[str]
display_order: int
show_in_footer: bool
show_in_header: bool
is_platform_default: bool
is_vendor_override: bool
created_at: str
updated_at: str
created_by: Optional[int]
updated_by: Optional[int]
# ============================================================================
# PLATFORM DEFAULT PAGES (vendor_id=NULL)
# ============================================================================
@router.get("/platform", response_model=List[ContentPageResponse])
def list_platform_pages(
include_unpublished: bool = Query(False, description="Include draft pages"),
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""
List all platform default content pages.
These are used as fallbacks when vendors haven't created custom pages.
"""
pages = content_page_service.list_all_platform_pages(
db,
include_unpublished=include_unpublished
)
return [page.to_dict() for page in pages]
@router.post("/platform", response_model=ContentPageResponse, status_code=201)
def create_platform_page(
page_data: ContentPageCreate,
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""
Create a new platform default content page.
Platform defaults are shown to all vendors who haven't created their own version.
"""
# Force vendor_id to None for platform pages
page = content_page_service.create_page(
db,
slug=page_data.slug,
title=page_data.title,
content=page_data.content,
vendor_id=None, # Platform default
content_format=page_data.content_format,
meta_description=page_data.meta_description,
meta_keywords=page_data.meta_keywords,
is_published=page_data.is_published,
show_in_footer=page_data.show_in_footer,
show_in_header=page_data.show_in_header,
display_order=page_data.display_order,
created_by=current_user.id
)
return page.to_dict()
# ============================================================================
# ALL CONTENT PAGES (Platform + Vendors)
# ============================================================================
@router.get("/", response_model=List[ContentPageResponse])
def list_all_pages(
vendor_id: Optional[int] = Query(None, description="Filter by vendor ID"),
include_unpublished: bool = Query(False, description="Include draft pages"),
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""
List all content pages (platform defaults and vendor overrides).
Filter by vendor_id to see specific vendor pages.
"""
if vendor_id:
pages = content_page_service.list_all_vendor_pages(
db,
vendor_id=vendor_id,
include_unpublished=include_unpublished
)
else:
# Get all pages (both platform and vendor)
from models.database.content_page import ContentPage
from sqlalchemy import and_
filters = []
if not include_unpublished:
filters.append(ContentPage.is_published == True)
pages = (
db.query(ContentPage)
.filter(and_(*filters) if filters else True)
.order_by(ContentPage.vendor_id, ContentPage.display_order, ContentPage.title)
.all()
)
return [page.to_dict() for page in pages]
@router.get("/{page_id}", response_model=ContentPageResponse)
def get_page(
page_id: int,
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""Get a specific content page by ID."""
page = content_page_service.get_page_by_id(db, page_id)
if not page:
raise HTTPException(status_code=404, detail="Content page not found")
return page.to_dict()
@router.put("/{page_id}", response_model=ContentPageResponse)
def update_page(
page_id: int,
page_data: ContentPageUpdate,
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""Update a content page (platform or vendor)."""
page = content_page_service.update_page(
db,
page_id=page_id,
title=page_data.title,
content=page_data.content,
content_format=page_data.content_format,
meta_description=page_data.meta_description,
meta_keywords=page_data.meta_keywords,
is_published=page_data.is_published,
show_in_footer=page_data.show_in_footer,
show_in_header=page_data.show_in_header,
display_order=page_data.display_order,
updated_by=current_user.id
)
if not page:
raise HTTPException(status_code=404, detail="Content page not found")
return page.to_dict()
@router.delete("/{page_id}", status_code=204)
def delete_page(
page_id: int,
current_user: User = Depends(get_current_admin_user),
db: Session = Depends(get_db)
):
"""Delete a content page."""
success = content_page_service.delete_page(db, page_id)
if not success:
raise HTTPException(status_code=404, detail="Content page not found")
return None

25
app/api/v1/shop/__init__.py Normal file
View File

@@ -0,0 +1,25 @@
# app/api/v1/shop/__init__.py
"""
Shop API router aggregation.
This module aggregates all shop-related JSON API endpoints (public facing).
These are PUBLIC endpoints - no authentication required.
"""
from fastapi import APIRouter
# Import shop routers
from . import content_pages
# Create shop router
router = APIRouter()
# ============================================================================
# PUBLIC API ROUTES
# ============================================================================
# Content pages (about, faq, contact, etc.)
router.include_router(content_pages.router, prefix="/content-pages", tags=["shop-content-pages"])
__all__ = ["router"]

116
app/api/v1/shop/content_pages.py Normal file
View File

@@ -0,0 +1,116 @@
# app/api/v1/shop/content_pages.py
"""
Shop Content Pages API (Public)
Public endpoints for retrieving content pages in shop frontend.
No authentication required.
"""
import logging
from typing import List
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.services.content_page_service import content_page_service
router = APIRouter()
logger = logging.getLogger(__name__)
# ============================================================================
# RESPONSE SCHEMAS
# ============================================================================
class PublicContentPageResponse(BaseModel):
"""Public content page response (no internal IDs)."""
slug: str
title: str
content: str
content_format: str
meta_description: str | None
meta_keywords: str | None
published_at: str | None
class ContentPageListItem(BaseModel):
"""Content page list item for navigation."""
slug: str
title: str
show_in_footer: bool
show_in_header: bool
display_order: int
# ============================================================================
# PUBLIC ENDPOINTS
# ============================================================================
@router.get("/navigation", response_model=List[ContentPageListItem])
def get_navigation_pages(
request: Request,
db: Session = Depends(get_db)
):
"""
Get list of content pages for navigation (footer/header).
Uses vendor from request.state (set by middleware).
Returns vendor overrides + platform defaults.
"""
vendor = getattr(request.state, 'vendor', None)
vendor_id = vendor.id if vendor else None
# Get all published pages for this vendor
pages = content_page_service.list_pages_for_vendor(
db,
vendor_id=vendor_id,
include_unpublished=False
)
return [
{
"slug": page.slug,
"title": page.title,
"show_in_footer": page.show_in_footer,
"show_in_header": page.show_in_header,
"display_order": page.display_order,
}
for page in pages
]
@router.get("/{slug}", response_model=PublicContentPageResponse)
def get_content_page(
slug: str,
request: Request,
db: Session = Depends(get_db)
):
"""
Get a specific content page by slug.
Uses vendor from request.state (set by middleware).
Returns vendor override if exists, otherwise platform default.
"""
vendor = getattr(request.state, 'vendor', None)
vendor_id = vendor.id if vendor else None
page = content_page_service.get_page_for_vendor(
db,
slug=slug,
vendor_id=vendor_id,
include_unpublished=False # Only show published pages
)
if not page:
raise HTTPException(status_code=404, detail=f"Content page not found: {slug}")
return {
"slug": page.slug,
"title": page.title,
"content": page.content,
"content_format": page.content_format,
"meta_description": page.meta_description,
"meta_keywords": page.meta_keywords,
"published_at": page.published_at.isoformat() if page.published_at else None,
}

4
app/api/v1/vendor/__init__.py vendored
View File

@@ -29,6 +29,7 @@ from . import (
media,
notifications,
analytics,
content_pages,
)
@@ -66,6 +67,9 @@ router.include_router(media.router, tags=["vendor-media"])
router.include_router(notifications.router, tags=["vendor-notifications"])
router.include_router(analytics.router, tags=["vendor-analytics"])
# Content pages management
router.include_router(content_pages.router, prefix="/{vendor_code}/content-pages", tags=["vendor-content-pages"])
# Vendor info endpoint - MUST BE LAST! Has catch-all GET /{vendor_code}
router.include_router(info.router, tags=["vendor-info"])

259
app/api/v1/vendor/content_pages.py vendored Normal file
View File

@@ -0,0 +1,259 @@
# app/api/v1/vendor/content_pages.py
"""
Vendor Content Pages API
Vendors can:
- View their content pages (includes platform defaults)
- Create/edit/delete their own content page overrides
- Preview pages before publishing
"""
import logging
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_vendor_user, get_db
from app.services.content_page_service import content_page_service
from models.database.user import User
router = APIRouter()
logger = logging.getLogger(__name__)
# ============================================================================
# REQUEST/RESPONSE SCHEMAS
# ============================================================================
class VendorContentPageCreate(BaseModel):
"""Schema for creating a vendor content page."""
slug: str = Field(..., max_length=100, description="URL-safe identifier (about, faq, contact, etc.)")
title: str = Field(..., max_length=200, description="Page title")
content: str = Field(..., description="HTML or Markdown content")
content_format: str = Field(default="html", description="Content format: html or markdown")
meta_description: Optional[str] = Field(None, max_length=300, description="SEO meta description")
meta_keywords: Optional[str] = Field(None, max_length=300, description="SEO keywords")
is_published: bool = Field(default=False, description="Publish immediately")
show_in_footer: bool = Field(default=True, description="Show in footer navigation")
show_in_header: bool = Field(default=False, description="Show in header navigation")
display_order: int = Field(default=0, description="Display order (lower = first)")
class VendorContentPageUpdate(BaseModel):
"""Schema for updating a vendor content page."""
title: Optional[str] = Field(None, max_length=200)
content: Optional[str] = None
content_format: Optional[str] = None
meta_description: Optional[str] = Field(None, max_length=300)
meta_keywords: Optional[str] = Field(None, max_length=300)
is_published: Optional[bool] = None
show_in_footer: Optional[bool] = None
show_in_header: Optional[bool] = None
display_order: Optional[int] = None
class ContentPageResponse(BaseModel):
"""Schema for content page response."""
id: int
vendor_id: Optional[int]
vendor_name: Optional[str]
slug: str
title: str
content: str
content_format: str
meta_description: Optional[str]
meta_keywords: Optional[str]
is_published: bool
published_at: Optional[str]
display_order: int
show_in_footer: bool
show_in_header: bool
is_platform_default: bool
is_vendor_override: bool
created_at: str
updated_at: str
created_by: Optional[int]
updated_by: Optional[int]
# ============================================================================
# VENDOR CONTENT PAGES
# ============================================================================
@router.get("/", response_model=List[ContentPageResponse])
def list_vendor_pages(
include_unpublished: bool = Query(False, description="Include draft pages"),
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
List all content pages available for this vendor.
Returns vendor-specific overrides + platform defaults (vendor overrides take precedence).
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
pages = content_page_service.list_pages_for_vendor(
db,
vendor_id=current_user.vendor_id,
include_unpublished=include_unpublished
)
return [page.to_dict() for page in pages]
@router.get("/overrides", response_model=List[ContentPageResponse])
def list_vendor_overrides(
include_unpublished: bool = Query(False, description="Include draft pages"),
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
List only vendor-specific content pages (no platform defaults).
Shows what the vendor has customized.
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
pages = content_page_service.list_all_vendor_pages(
db,
vendor_id=current_user.vendor_id,
include_unpublished=include_unpublished
)
return [page.to_dict() for page in pages]
@router.get("/{slug}", response_model=ContentPageResponse)
def get_page(
slug: str,
include_unpublished: bool = Query(False, description="Include draft pages"),
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
Get a specific content page by slug.
Returns vendor override if exists, otherwise platform default.
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
page = content_page_service.get_page_for_vendor(
db,
slug=slug,
vendor_id=current_user.vendor_id,
include_unpublished=include_unpublished
)
if not page:
raise HTTPException(status_code=404, detail=f"Content page not found: {slug}")
return page.to_dict()
@router.post("/", response_model=ContentPageResponse, status_code=201)
def create_vendor_page(
page_data: VendorContentPageCreate,
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
Create a vendor-specific content page override.
This will be shown instead of the platform default for this vendor.
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
page = content_page_service.create_page(
db,
slug=page_data.slug,
title=page_data.title,
content=page_data.content,
vendor_id=current_user.vendor_id,
content_format=page_data.content_format,
meta_description=page_data.meta_description,
meta_keywords=page_data.meta_keywords,
is_published=page_data.is_published,
show_in_footer=page_data.show_in_footer,
show_in_header=page_data.show_in_header,
display_order=page_data.display_order,
created_by=current_user.id
)
return page.to_dict()
@router.put("/{page_id}", response_model=ContentPageResponse)
def update_vendor_page(
page_id: int,
page_data: VendorContentPageUpdate,
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
Update a vendor-specific content page.
Can only update pages owned by this vendor.
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
# Verify ownership
existing_page = content_page_service.get_page_by_id(db, page_id)
if not existing_page:
raise HTTPException(status_code=404, detail="Content page not found")
if existing_page.vendor_id != current_user.vendor_id:
raise HTTPException(status_code=403, detail="Cannot edit pages from other vendors")
# Update
page = content_page_service.update_page(
db,
page_id=page_id,
title=page_data.title,
content=page_data.content,
content_format=page_data.content_format,
meta_description=page_data.meta_description,
meta_keywords=page_data.meta_keywords,
is_published=page_data.is_published,
show_in_footer=page_data.show_in_footer,
show_in_header=page_data.show_in_header,
display_order=page_data.display_order,
updated_by=current_user.id
)
return page.to_dict()
@router.delete("/{page_id}", status_code=204)
def delete_vendor_page(
page_id: int,
current_user: User = Depends(get_current_vendor_user),
db: Session = Depends(get_db)
):
"""
Delete a vendor-specific content page.
Can only delete pages owned by this vendor.
After deletion, platform default will be shown (if exists).
"""
if not current_user.vendor_id:
raise HTTPException(status_code=403, detail="User is not associated with a vendor")
# Verify ownership
existing_page = content_page_service.get_page_by_id(db, page_id)
if not existing_page:
raise HTTPException(status_code=404, detail="Content page not found")
if existing_page.vendor_id != current_user.vendor_id:
raise HTTPException(status_code=403, detail="Cannot delete pages from other vendors")
# Delete
content_page_service.delete_page(db, page_id)
return None

390
app/services/content_page_service.py Normal file
View File

@@ -0,0 +1,390 @@
# app/services/content_page_service.py
"""
Content Page Service
Business logic for managing content pages with platform defaults
and vendor-specific overrides.
Lookup Strategy:
1. Check for vendor-specific override (vendor_id + slug + published)
2. If not found, check for platform default (slug + published)
3. If neither exists, return None
This allows:
- Platform admin to create default content for all vendors
- Vendors to override specific pages with custom content
- Fallback to platform defaults when vendor hasn't customized
"""
import logging
from datetime import datetime, timezone
from typing import List, Optional
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from models.database.content_page import ContentPage
logger = logging.getLogger(__name__)
class ContentPageService:
"""Service for content page operations."""
@staticmethod
def get_page_for_vendor(
db: Session,
slug: str,
vendor_id: Optional[int] = None,
include_unpublished: bool = False
) -> Optional[ContentPage]:
"""
Get content page for a vendor with fallback to platform default.
Lookup order:
1. Vendor-specific override (vendor_id + slug)
2. Platform default (vendor_id=NULL + slug)
Args:
db: Database session
slug: Page slug (about, faq, contact, etc.)
vendor_id: Vendor ID (None for platform defaults only)
include_unpublished: Include draft pages (for preview)
Returns:
ContentPage or None
"""
filters = [ContentPage.slug == slug]
if not include_unpublished:
filters.append(ContentPage.is_published == True)
# Try vendor-specific override first
if vendor_id:
vendor_page = (
db.query(ContentPage)
.filter(
and_(
ContentPage.vendor_id == vendor_id,
*filters
)
)
.first()
)
if vendor_page:
logger.debug(f"Found vendor-specific page: {slug} for vendor_id={vendor_id}")
return vendor_page
# Fallback to platform default
platform_page = (
db.query(ContentPage)
.filter(
and_(
ContentPage.vendor_id == None,
*filters
)
)
.first()
)
if platform_page:
logger.debug(f"Using platform default page: {slug}")
else:
logger.debug(f"No page found for slug: {slug}")
return platform_page
@staticmethod
def list_pages_for_vendor(
db: Session,
vendor_id: Optional[int] = None,
include_unpublished: bool = False,
footer_only: bool = False,
header_only: bool = False
) -> List[ContentPage]:
"""
List all available pages for a vendor (includes vendor overrides + platform defaults).
Merges vendor-specific overrides with platform defaults, prioritizing vendor overrides.
Args:
db: Database session
vendor_id: Vendor ID (None for platform pages only)
include_unpublished: Include draft pages
footer_only: Only pages marked for footer display
header_only: Only pages marked for header display
Returns:
List of ContentPage objects
"""
filters = []
if not include_unpublished:
filters.append(ContentPage.is_published == True)
if footer_only:
filters.append(ContentPage.show_in_footer == True)
if header_only:
filters.append(ContentPage.show_in_header == True)
# Get vendor-specific pages
vendor_pages = []
if vendor_id:
vendor_pages = (
db.query(ContentPage)
.filter(
and_(
ContentPage.vendor_id == vendor_id,
*filters
)
)
.order_by(ContentPage.display_order, ContentPage.title)
.all()
)
# Get platform defaults
platform_pages = (
db.query(ContentPage)
.filter(
and_(
ContentPage.vendor_id == None,
*filters
)
)
.order_by(ContentPage.display_order, ContentPage.title)
.all()
)
# Merge: vendor overrides take precedence
vendor_slugs = {page.slug for page in vendor_pages}
all_pages = vendor_pages + [
page for page in platform_pages
if page.slug not in vendor_slugs
]
# Sort by display_order
all_pages.sort(key=lambda p: (p.display_order, p.title))
return all_pages
@staticmethod
def create_page(
db: Session,
slug: str,
title: str,
content: str,
vendor_id: Optional[int] = None,
content_format: str = "html",
meta_description: Optional[str] = None,
meta_keywords: Optional[str] = None,
is_published: bool = False,
show_in_footer: bool = True,
show_in_header: bool = False,
display_order: int = 0,
created_by: Optional[int] = None
) -> ContentPage:
"""
Create a new content page.
Args:
db: Database session
slug: URL-safe identifier
title: Page title
content: HTML or Markdown content
vendor_id: Vendor ID (None for platform default)
content_format: "html" or "markdown"
meta_description: SEO description
meta_keywords: SEO keywords
is_published: Publish immediately
show_in_footer: Show in footer navigation
show_in_header: Show in header navigation
display_order: Sort order
created_by: User ID who created it
Returns:
Created ContentPage
"""
page = ContentPage(
vendor_id=vendor_id,
slug=slug,
title=title,
content=content,
content_format=content_format,
meta_description=meta_description,
meta_keywords=meta_keywords,
is_published=is_published,
published_at=datetime.now(timezone.utc) if is_published else None,
show_in_footer=show_in_footer,
show_in_header=show_in_header,
display_order=display_order,
created_by=created_by,
updated_by=created_by,
)
db.add(page)
db.commit()
db.refresh(page)
logger.info(f"Created content page: {slug} (vendor_id={vendor_id}, id={page.id})")
return page
@staticmethod
def update_page(
db: Session,
page_id: int,
title: Optional[str] = None,
content: Optional[str] = None,
content_format: Optional[str] = None,
meta_description: Optional[str] = None,
meta_keywords: Optional[str] = None,
is_published: Optional[bool] = None,
show_in_footer: Optional[bool] = None,
show_in_header: Optional[bool] = None,
display_order: Optional[int] = None,
updated_by: Optional[int] = None
) -> Optional[ContentPage]:
"""
Update an existing content page.
Args:
db: Database session
page_id: Page ID
title: New title
content: New content
content_format: New format
meta_description: New SEO description
meta_keywords: New SEO keywords
is_published: New publish status
show_in_footer: New footer visibility
show_in_header: New header visibility
display_order: New sort order
updated_by: User ID who updated it
Returns:
Updated ContentPage or None if not found
"""
page = db.query(ContentPage).filter(ContentPage.id == page_id).first()
if not page:
logger.warning(f"Content page not found: id={page_id}")
return None
# Update fields if provided
if title is not None:
page.title = title
if content is not None:
page.content = content
if content_format is not None:
page.content_format = content_format
if meta_description is not None:
page.meta_description = meta_description
if meta_keywords is not None:
page.meta_keywords = meta_keywords
if is_published is not None:
page.is_published = is_published
if is_published and not page.published_at:
page.published_at = datetime.now(timezone.utc)
if show_in_footer is not None:
page.show_in_footer = show_in_footer
if show_in_header is not None:
page.show_in_header = show_in_header
if display_order is not None:
page.display_order = display_order
if updated_by is not None:
page.updated_by = updated_by
db.commit()
db.refresh(page)
logger.info(f"Updated content page: id={page_id}, slug={page.slug}")
return page
@staticmethod
def delete_page(db: Session, page_id: int) -> bool:
"""
Delete a content page.
Args:
db: Database session
page_id: Page ID
Returns:
True if deleted, False if not found
"""
page = db.query(ContentPage).filter(ContentPage.id == page_id).first()
if not page:
logger.warning(f"Content page not found for deletion: id={page_id}")
return False
db.delete(page)
db.commit()
logger.info(f"Deleted content page: id={page_id}, slug={page.slug}")
return True
@staticmethod
def get_page_by_id(db: Session, page_id: int) -> Optional[ContentPage]:
"""Get content page by ID."""
return db.query(ContentPage).filter(ContentPage.id == page_id).first()
@staticmethod
def list_all_vendor_pages(
db: Session,
vendor_id: int,
include_unpublished: bool = False
) -> List[ContentPage]:
"""
List only vendor-specific pages (no platform defaults).
Args:
db: Database session
vendor_id: Vendor ID
include_unpublished: Include draft pages
Returns:
List of vendor-specific ContentPage objects
"""
filters = [ContentPage.vendor_id == vendor_id]
if not include_unpublished:
filters.append(ContentPage.is_published == True)
return (
db.query(ContentPage)
.filter(and_(*filters))
.order_by(ContentPage.display_order, ContentPage.title)
.all()
)
@staticmethod
def list_all_platform_pages(
db: Session,
include_unpublished: bool = False
) -> List[ContentPage]:
"""
List only platform default pages.
Args:
db: Database session
include_unpublished: Include draft pages
Returns:
List of platform default ContentPage objects
"""
filters = [ContentPage.vendor_id == None]
if not include_unpublished:
filters.append(ContentPage.is_published == True)
return (
db.query(ContentPage)
.filter(and_(*filters))
.order_by(ContentPage.display_order, ContentPage.title)
.all()
)
# Singleton instance
content_page_service = ContentPageService()