Files
orion/app/api/v1/admin/platforms.py
Samir Boulahtit cad862f469 refactor(api): introduce UserContext schema for API dependency injection
Replace direct User database model imports in API endpoints with UserContext
schema, following the architecture principle that API routes should not import
database models directly.

Changes:
- Create UserContext schema in models/schema/auth.py with from_user() factory
- Update app/api/deps.py to return UserContext from all auth dependencies
- Add _get_user_model() helper for functions needing User model access
- Update 58 API endpoint files to use UserContext instead of User
- Add noqa comments for 4 legitimate edge cases (enums, internal helpers)

Architecture validation: 0 errors (down from 61), 11 warnings remain

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-30 20:47:33 +01:00

225 lines
7.3 KiB
Python

# app/api/v1/admin/platforms.py
"""
Admin API endpoints for Platform management (Multi-Platform CMS).
Provides CRUD operations for platforms:
- GET /platforms - List all platforms
- GET /platforms/{code} - Get platform details
- PUT /platforms/{code} - Update platform settings
- GET /platforms/{code}/stats - Get platform statistics
Platforms are business offerings (OMS, Loyalty, Site Builder) with their own:
- Marketing pages (homepage, pricing, features)
- Vendor defaults (about, terms, privacy)
- Configuration and branding
"""
import logging
from typing import Any
from fastapi import APIRouter, Depends, Path, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_from_cookie_or_header, get_db
from app.services.platform_service import platform_service
from models.schema.auth import UserContext
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/platforms")
# =============================================================================
# Pydantic Schemas
# =============================================================================
class PlatformResponse(BaseModel):
"""Platform response schema."""
id: int
code: str
name: str
description: str | None = None
domain: str | None = None
path_prefix: str | None = None
logo: str | None = None
logo_dark: str | None = None
favicon: str | None = None
theme_config: dict[str, Any] = Field(default_factory=dict)
default_language: str = "fr"
supported_languages: list[str] = Field(default_factory=lambda: ["fr", "de", "en"])
is_active: bool = True
is_public: bool = True
settings: dict[str, Any] = Field(default_factory=dict)
created_at: str
updated_at: str
# Computed fields (added by endpoint)
vendor_count: int = 0
platform_pages_count: int = 0
vendor_defaults_count: int = 0
class Config:
from_attributes = True
class PlatformListResponse(BaseModel):
"""Response for platform list."""
platforms: list[PlatformResponse]
total: int
class PlatformUpdateRequest(BaseModel):
"""Request schema for updating a platform."""
name: str | None = None
description: str | None = None
domain: str | None = None
path_prefix: str | None = None
logo: str | None = None
logo_dark: str | None = None
favicon: str | None = None
theme_config: dict[str, Any] | None = None
default_language: str | None = None
supported_languages: list[str] | None = None
is_active: bool | None = None
is_public: bool | None = None
settings: dict[str, Any] | None = None
class PlatformStatsResponse(BaseModel):
"""Platform statistics response."""
platform_id: int
platform_code: str
platform_name: str
vendor_count: int
platform_pages_count: int
vendor_defaults_count: int
vendor_overrides_count: int
published_pages_count: int
draft_pages_count: int
# =============================================================================
# Helper Functions
# =============================================================================
def _build_platform_response(db: Session, platform) -> PlatformResponse:
"""Build PlatformResponse from Platform model with computed fields."""
return PlatformResponse(
id=platform.id,
code=platform.code,
name=platform.name,
description=platform.description,
domain=platform.domain,
path_prefix=platform.path_prefix,
logo=platform.logo,
logo_dark=platform.logo_dark,
favicon=platform.favicon,
theme_config=platform.theme_config or {},
default_language=platform.default_language,
supported_languages=platform.supported_languages or ["fr", "de", "en"],
is_active=platform.is_active,
is_public=platform.is_public,
settings=platform.settings or {},
created_at=platform.created_at.isoformat(),
updated_at=platform.updated_at.isoformat(),
vendor_count=platform_service.get_vendor_count(db, platform.id),
platform_pages_count=platform_service.get_platform_pages_count(db, platform.id),
vendor_defaults_count=platform_service.get_vendor_defaults_count(db, platform.id),
)
# =============================================================================
# API Endpoints
# =============================================================================
@router.get("", response_model=PlatformListResponse)
async def list_platforms(
db: Session = Depends(get_db),
current_user: UserContext = Depends(get_current_admin_from_cookie_or_header),
include_inactive: bool = Query(False, description="Include inactive platforms"),
):
"""
List all platforms with their statistics.
Returns all platforms (OMS, Loyalty, etc.) with vendor counts and page counts.
"""
platforms = platform_service.list_platforms(db, include_inactive=include_inactive)
result = [_build_platform_response(db, platform) for platform in platforms]
logger.info(f"[PLATFORMS] Listed {len(result)} platforms")
return PlatformListResponse(platforms=result, total=len(result))
@router.get("/{code}", response_model=PlatformResponse)
async def get_platform(
code: str = Path(..., description="Platform code (oms, loyalty, etc.)"),
db: Session = Depends(get_db),
current_user: UserContext = Depends(get_current_admin_from_cookie_or_header),
):
"""
Get platform details by code.
Returns full platform configuration including statistics.
"""
platform = platform_service.get_platform_by_code(db, code)
return _build_platform_response(db, platform)
@router.put("/{code}", response_model=PlatformResponse)
async def update_platform(
update_data: PlatformUpdateRequest,
code: str = Path(..., description="Platform code"),
db: Session = Depends(get_db),
current_user: UserContext = Depends(get_current_admin_from_cookie_or_header),
):
"""
Update platform settings.
Allows updating name, description, branding, and configuration.
"""
platform = platform_service.get_platform_by_code(db, code)
update_dict = update_data.model_dump(exclude_unset=True)
platform = platform_service.update_platform(db, platform, update_dict)
db.commit()
db.refresh(platform)
return _build_platform_response(db, platform)
@router.get("/{code}/stats", response_model=PlatformStatsResponse)
async def get_platform_stats(
code: str = Path(..., description="Platform code"),
db: Session = Depends(get_db),
current_user: UserContext = Depends(get_current_admin_from_cookie_or_header),
):
"""
Get detailed statistics for a platform.
Returns counts for vendors, pages, and content breakdown.
"""
platform = platform_service.get_platform_by_code(db, code)
stats = platform_service.get_platform_stats(db, platform)
return PlatformStatsResponse(
platform_id=stats.platform_id,
platform_code=stats.platform_code,
platform_name=stats.platform_name,
vendor_count=stats.vendor_count,
platform_pages_count=stats.platform_pages_count,
vendor_defaults_count=stats.vendor_defaults_count,
vendor_overrides_count=stats.vendor_overrides_count,
published_pages_count=stats.published_pages_count,
draft_pages_count=stats.draft_pages_count,
)