From 2ce19e66b1ab2ccdab476b633e11dd153b120a46 Mon Sep 17 00:00:00 2001 From: Samir Boulahtit Date: Mon, 26 Jan 2026 21:35:36 +0100 Subject: [PATCH] feat: implement self-contained module architecture (Phase 1 & 2) Phase 1 - Foundation: - Add app/modules/contracts/ with Protocol definitions for cross-module communication (ServiceProtocol, ContentServiceProtocol, MediaServiceProtocol) - Enhance app/modules/base.py ModuleDefinition with self-contained module support (is_self_contained, services_path, models_path, etc.) - Update app/templates_config.py with multi-directory template loading using Jinja2 ChoiceLoader for module templates Phase 2 - CMS Pilot Module: - Migrate CMS service to app/modules/cms/services/content_page_service.py - Create app/modules/cms/exceptions.py with CMS-specific exceptions - Configure app/modules/cms/models/ to re-export ContentPage from canonical location (models.database) to avoid circular imports - Update cms_module definition with is_self_contained=True and paths - Add backwards compatibility shims with deprecation warnings: - app/services/content_page_service.py -> app.modules.cms.services - app/exceptions/content_page.py -> app.modules.cms.exceptions Note: SQLAlchemy models remain in models/database/ as the canonical location to avoid circular imports at startup time. Module model packages re-export from the canonical location. Co-Authored-By: Claude Opus 4.5 --- app/exceptions/content_page.py | 115 +- app/modules/base.py | 140 ++- app/modules/cms/__init__.py | 17 +- app/modules/cms/definition.py | 19 +- app/modules/cms/exceptions.py | 110 ++ app/modules/cms/models/__init__.py | 20 + app/modules/cms/services/__init__.py | 16 + .../cms/services/content_page_service.py | 1002 ++++++++++++++++ app/modules/contracts/__init__.py | 32 + app/modules/contracts/base.py | 47 + app/modules/contracts/cms.py | 148 +++ app/services/content_page_service.py | 1009 +---------------- app/templates_config.py | 66 +- models/database/content_page.py | 8 +- 14 files changed, 1663 insertions(+), 1086 deletions(-) create mode 100644 app/modules/cms/exceptions.py create mode 100644 app/modules/cms/models/__init__.py create mode 100644 app/modules/cms/services/__init__.py create mode 100644 app/modules/cms/services/content_page_service.py create mode 100644 app/modules/contracts/__init__.py create mode 100644 app/modules/contracts/base.py create mode 100644 app/modules/contracts/cms.py diff --git a/app/exceptions/content_page.py b/app/exceptions/content_page.py index 300c8342..3cbd10ee 100644 --- a/app/exceptions/content_page.py +++ b/app/exceptions/content_page.py @@ -1,88 +1,43 @@ # app/exceptions/content_page.py """ -Content Page Domain Exceptions +DEPRECATED: This module has moved to app.modules.cms.exceptions -These exceptions are raised by the content page service layer -and converted to HTTP responses by the global exception handler. +Please update your imports: + # Old (deprecated): + from app.exceptions.content_page import ContentPageNotFoundException + + # New (preferred): + from app.modules.cms.exceptions import ContentPageNotFoundException + +This shim re-exports from the new location for backwards compatibility. """ -from app.exceptions.base import ( - AuthorizationException, - BusinessLogicException, - ConflictException, - ResourceNotFoundException, - ValidationException, +import warnings + +warnings.warn( + "Import from app.modules.cms.exceptions instead of " + "app.exceptions.content_page. This shim will be removed in a future version.", + DeprecationWarning, + stacklevel=2, ) +# Re-export everything from the new location +from app.modules.cms.exceptions import ( # noqa: E402, F401 + ContentPageAlreadyExistsException, + ContentPageNotFoundException, + ContentPageNotPublishedException, + ContentPageSlugReservedException, + ContentPageValidationException, + UnauthorizedContentPageAccessException, + VendorNotAssociatedException, +) -class ContentPageNotFoundException(ResourceNotFoundException): - """Raised when a content page is not found.""" - - def __init__(self, identifier: str | int | None = None): - if identifier: - message = f"Content page not found: {identifier}" - else: - message = "Content page not found" - super().__init__( - message=message, - resource_type="content_page", - identifier=str(identifier) if identifier else "unknown", - ) - - -class ContentPageAlreadyExistsException(ConflictException): - """Raised when a content page with the same slug already exists.""" - - def __init__(self, slug: str, vendor_id: int | None = None): - if vendor_id: - message = f"Content page with slug '{slug}' already exists for this vendor" - else: - message = f"Platform content page with slug '{slug}' already exists" - super().__init__(message=message) - - -class ContentPageSlugReservedException(ValidationException): - """Raised when trying to use a reserved slug.""" - - def __init__(self, slug: str): - super().__init__( - message=f"Content page slug '{slug}' is reserved", - field="slug", - value=slug, - ) - - -class ContentPageNotPublishedException(BusinessLogicException): - """Raised when trying to access an unpublished content page.""" - - def __init__(self, slug: str): - super().__init__(message=f"Content page '{slug}' is not published") - - -class UnauthorizedContentPageAccessException(AuthorizationException): - """Raised when a user tries to access/modify a content page they don't own.""" - - def __init__(self, action: str = "access"): - super().__init__( - message=f"Cannot {action} content pages from other vendors", - error_code="CONTENT_PAGE_ACCESS_DENIED", - details={"required_permission": f"content_page:{action}"}, - ) - - -class VendorNotAssociatedException(AuthorizationException): - """Raised when a user is not associated with a vendor.""" - - def __init__(self): - super().__init__( - message="User is not associated with a vendor", - error_code="VENDOR_NOT_ASSOCIATED", - details={"required_permission": "vendor:member"}, - ) - - -class ContentPageValidationException(ValidationException): - """Raised when content page data validation fails.""" - - def __init__(self, field: str, message: str, value: str | None = None): - super().__init__(message=message, field=field, value=value) +__all__ = [ + "ContentPageNotFoundException", + "ContentPageAlreadyExistsException", + "ContentPageSlugReservedException", + "ContentPageNotPublishedException", + "UnauthorizedContentPageAccessException", + "VendorNotAssociatedException", + "ContentPageValidationException", +] diff --git a/app/modules/base.py b/app/modules/base.py index fc13c549..74a8a05a 100644 --- a/app/modules/base.py +++ b/app/modules/base.py @@ -7,9 +7,26 @@ per platform. Each module contains: - Features: Granular capabilities for tier-based access control - Menu items: Sidebar entries per frontend type - Routes: API and page routes (future: dynamically registered) +- Services: Business logic (optional, for self-contained modules) +- Models: Database models (optional, for self-contained modules) +- Schemas: Pydantic schemas (optional, for self-contained modules) +- Templates: Jinja2 templates (optional, for self-contained modules) + +Self-Contained Module Structure: + app/modules// + ├── __init__.py + ├── definition.py # ModuleDefinition instance + ├── config.py # Module configuration schema (optional) + ├── exceptions.py # Module-specific exceptions (optional) + ├── routes/ # FastAPI routers + ├── services/ # Business logic + ├── models/ # SQLAlchemy models + ├── schemas/ # Pydantic schemas + └── templates/ # Jinja2 templates (namespaced) """ from dataclasses import dataclass, field +from pathlib import Path from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -26,6 +43,9 @@ class ModuleDefinition: A module groups related functionality that can be enabled/disabled per platform. Core modules cannot be disabled and are always available. + Self-contained modules include their own services, models, schemas, and templates. + The path attributes describe where these components are located. + Attributes: code: Unique identifier (e.g., "billing", "marketplace") name: Display name (e.g., "Billing & Subscriptions") @@ -34,10 +54,16 @@ class ModuleDefinition: features: List of feature codes this module provides menu_items: Dict mapping FrontendType to list of menu item IDs is_core: Core modules cannot be disabled - admin_router: FastAPI router for admin routes (future) - vendor_router: FastAPI router for vendor routes (future) + admin_router: FastAPI router for admin routes + vendor_router: FastAPI router for vendor routes + services_path: Path to services subpackage (e.g., "app.modules.billing.services") + models_path: Path to models subpackage (e.g., "app.modules.billing.models") + schemas_path: Path to schemas subpackage (e.g., "app.modules.billing.schemas") + templates_path: Path to templates directory (relative to module) + exceptions_path: Path to exceptions module (e.g., "app.modules.billing.exceptions") + is_self_contained: Whether module uses self-contained structure - Example: + Example (traditional thin wrapper): billing_module = ModuleDefinition( code="billing", name="Billing & Subscriptions", @@ -48,6 +74,24 @@ class ModuleDefinition: FrontendType.VENDOR: ["billing"], }, ) + + Example (self-contained module): + cms_module = ModuleDefinition( + code="cms", + name="Content Management", + description="Content pages, media library, and vendor themes.", + features=["cms_basic", "cms_custom_pages"], + menu_items={ + FrontendType.ADMIN: ["content-pages"], + FrontendType.VENDOR: ["content-pages", "media"], + }, + is_self_contained=True, + services_path="app.modules.cms.services", + models_path="app.modules.cms.models", + schemas_path="app.modules.cms.schemas", + templates_path="templates", + exceptions_path="app.modules.cms.exceptions", + ) """ # Identity @@ -65,10 +109,18 @@ class ModuleDefinition: # Status is_core: bool = False - # Routes (registered dynamically) - Future implementation + # Routes (registered dynamically) admin_router: "APIRouter | None" = None vendor_router: "APIRouter | None" = None + # Self-contained module paths (optional) + is_self_contained: bool = False + services_path: str | None = None + models_path: str | None = None + schemas_path: str | None = None + templates_path: str | None = None # Relative to module directory + exceptions_path: str | None = None + def get_menu_items(self, frontend_type: FrontendType) -> list[str]: """Get menu item IDs for a specific frontend type.""" return self.menu_items.get(frontend_type, []) @@ -100,6 +152,83 @@ class ModuleDefinition: """ return [req for req in self.requires if req not in enabled_modules] + # ========================================================================= + # Self-Contained Module Methods + # ========================================================================= + + def get_module_dir(self) -> Path: + """ + Get the filesystem path to this module's directory. + + Returns: + Path to app/modules// + """ + return Path(__file__).parent / self.code + + def get_templates_dir(self) -> Path | None: + """ + Get the filesystem path to this module's templates directory. + + Returns: + Path to templates directory, or None if not self-contained + """ + if not self.is_self_contained or not self.templates_path: + return None + return self.get_module_dir() / self.templates_path + + def get_import_path(self, component: str) -> str | None: + """ + Get the Python import path for a module component. + + Args: + component: One of "services", "models", "schemas", "exceptions" + + Returns: + Import path string, or None if not configured + """ + paths = { + "services": self.services_path, + "models": self.models_path, + "schemas": self.schemas_path, + "exceptions": self.exceptions_path, + } + return paths.get(component) + + def validate_structure(self) -> list[str]: + """ + Validate that self-contained module has expected directory structure. + + Returns: + List of missing or invalid paths (empty if valid) + """ + if not self.is_self_contained: + return [] + + issues = [] + module_dir = self.get_module_dir() + + if not module_dir.exists(): + issues.append(f"Module directory not found: {module_dir}") + return issues + + # Check optional directories + expected_dirs = [] + if self.services_path: + expected_dirs.append("services") + if self.models_path: + expected_dirs.append("models") + if self.schemas_path: + expected_dirs.append("schemas") + if self.templates_path: + expected_dirs.append(self.templates_path) + + for dir_name in expected_dirs: + dir_path = module_dir / dir_name + if not dir_path.exists(): + issues.append(f"Missing directory: {dir_path}") + + return issues + def __hash__(self) -> int: return hash(self.code) @@ -109,4 +238,5 @@ class ModuleDefinition: return False def __repr__(self) -> str: - return f"" + sc = ", self_contained" if self.is_self_contained else "" + return f"" diff --git a/app/modules/cms/__init__.py b/app/modules/cms/__init__.py index 39eb2583..f0d44725 100644 --- a/app/modules/cms/__init__.py +++ b/app/modules/cms/__init__.py @@ -2,8 +2,13 @@ """ CMS Module - Content Management System. +This is a SELF-CONTAINED module that includes: +- Services: content_page_service (business logic) +- Models: ContentPage (database model) +- Exceptions: ContentPageNotFoundException, etc. + This module provides: -- Content pages management +- Content pages management (three-tier: platform, vendor default, vendor override) - Media library - Vendor themes - SEO tools @@ -15,6 +20,16 @@ Routes: Menu Items: - Admin: content-pages, vendor-themes - Vendor: content-pages, media + +Usage: + # Preferred: Import from module directly + from app.modules.cms.services import content_page_service + from app.modules.cms.models import ContentPage + from app.modules.cms.exceptions import ContentPageNotFoundException + + # Legacy: Still works via re-export shims (deprecated) + from app.services.content_page_service import content_page_service + from models.database.content_page import ContentPage """ from app.modules.cms.definition import cms_module diff --git a/app/modules/cms/definition.py b/app/modules/cms/definition.py index 8fef7877..aa60635b 100644 --- a/app/modules/cms/definition.py +++ b/app/modules/cms/definition.py @@ -3,7 +3,15 @@ CMS module definition. Defines the CMS module including its features, menu items, -and route configurations. +route configurations, and self-contained component paths. + +This is a self-contained module with: +- Services: app.modules.cms.services +- Models: app.modules.cms.models +- Exceptions: app.modules.cms.exceptions + +Templates remain in core (app/templates/admin/) for now due to +admin/base.html inheritance dependency. """ from app.modules.base import ModuleDefinition @@ -24,7 +32,7 @@ def _get_vendor_router(): return vendor_router -# CMS module definition +# CMS module definition - Self-contained module (pilot) cms_module = ModuleDefinition( code="cms", name="Content Management", @@ -48,6 +56,13 @@ cms_module = ModuleDefinition( ], }, is_core=False, + # Self-contained module configuration + is_self_contained=True, + services_path="app.modules.cms.services", + models_path="app.modules.cms.models", + exceptions_path="app.modules.cms.exceptions", + # Templates remain in core for now (admin/content-pages*.html) + templates_path=None, ) diff --git a/app/modules/cms/exceptions.py b/app/modules/cms/exceptions.py new file mode 100644 index 00000000..e962f62a --- /dev/null +++ b/app/modules/cms/exceptions.py @@ -0,0 +1,110 @@ +# app/modules/cms/exceptions.py +""" +CMS Module Exceptions + +These exceptions are raised by the CMS module service layer +and converted to HTTP responses by the global exception handler. +""" + +from app.exceptions.base import ( + AuthorizationException, + BusinessLogicException, + ConflictException, + ResourceNotFoundException, + ValidationException, +) + + +class ContentPageNotFoundException(ResourceNotFoundException): + """Raised when a content page is not found.""" + + def __init__(self, identifier: str | int | None = None): + if identifier: + message = f"Content page not found: {identifier}" + else: + message = "Content page not found" + super().__init__( + message=message, + resource_type="content_page", + identifier=str(identifier) if identifier else "unknown", + ) + + +class ContentPageAlreadyExistsException(ConflictException): + """Raised when a content page with the same slug already exists.""" + + def __init__(self, slug: str, vendor_id: int | None = None): + if vendor_id: + message = f"Content page with slug '{slug}' already exists for this vendor" + else: + message = f"Platform content page with slug '{slug}' already exists" + super().__init__(message=message) + + +class ContentPageSlugReservedException(ValidationException): + """Raised when trying to use a reserved slug.""" + + def __init__(self, slug: str): + super().__init__( + message=f"Content page slug '{slug}' is reserved", + field="slug", + value=slug, + ) + + +class ContentPageNotPublishedException(BusinessLogicException): + """Raised when trying to access an unpublished content page.""" + + def __init__(self, slug: str): + super().__init__(message=f"Content page '{slug}' is not published") + + +class UnauthorizedContentPageAccessException(AuthorizationException): + """Raised when a user tries to access/modify a content page they don't own.""" + + def __init__(self, action: str = "access"): + super().__init__( + message=f"Cannot {action} content pages from other vendors", + error_code="CONTENT_PAGE_ACCESS_DENIED", + details={"required_permission": f"content_page:{action}"}, + ) + + +class VendorNotAssociatedException(AuthorizationException): + """Raised when a user is not associated with a vendor.""" + + def __init__(self): + super().__init__( + message="User is not associated with a vendor", + error_code="VENDOR_NOT_ASSOCIATED", + details={"required_permission": "vendor:member"}, + ) + + +class ContentPageValidationException(ValidationException): + """Raised when content page data validation fails.""" + + def __init__(self, field: str, message: str, value: str | None = None): + super().__init__(message=message, field=field, value=value) + + +class MediaNotFoundException(ResourceNotFoundException): + """Raised when a media item is not found.""" + + def __init__(self, identifier: str | int | None = None): + if identifier: + message = f"Media item not found: {identifier}" + else: + message = "Media item not found" + super().__init__( + message=message, + resource_type="media", + identifier=str(identifier) if identifier else "unknown", + ) + + +class MediaUploadException(BusinessLogicException): + """Raised when a media upload fails.""" + + def __init__(self, reason: str): + super().__init__(message=f"Media upload failed: {reason}") diff --git a/app/modules/cms/models/__init__.py b/app/modules/cms/models/__init__.py new file mode 100644 index 00000000..60b51aac --- /dev/null +++ b/app/modules/cms/models/__init__.py @@ -0,0 +1,20 @@ +# app/modules/cms/models/__init__.py +""" +CMS module database models. + +This package re-exports the ContentPage model from its canonical location +in models.database. SQLAlchemy models must remain in a single location to +avoid circular imports at startup time. + +Usage: + from app.modules.cms.models import ContentPage + +The canonical model is at: models.database.content_page.ContentPage +""" + +# Import from canonical location to avoid circular imports +from models.database.content_page import ContentPage + +__all__ = [ + "ContentPage", +] diff --git a/app/modules/cms/services/__init__.py b/app/modules/cms/services/__init__.py new file mode 100644 index 00000000..1b2d2ab6 --- /dev/null +++ b/app/modules/cms/services/__init__.py @@ -0,0 +1,16 @@ +# app/modules/cms/services/__init__.py +""" +CMS module services. + +This package contains all business logic for the CMS module. +""" + +from app.modules.cms.services.content_page_service import ( + ContentPageService, + content_page_service, +) + +__all__ = [ + "ContentPageService", + "content_page_service", +] diff --git a/app/modules/cms/services/content_page_service.py b/app/modules/cms/services/content_page_service.py new file mode 100644 index 00000000..ebd5ff7e --- /dev/null +++ b/app/modules/cms/services/content_page_service.py @@ -0,0 +1,1002 @@ +# app/modules/cms/services/content_page_service.py +""" +Content Page Service + +Business logic for managing content pages with three-tier hierarchy: + +1. Platform Marketing Pages (is_platform_page=True, vendor_id=NULL) + - Platform's own pages (homepage, pricing, about) + - Describe the platform/business offering itself + +2. Vendor Default Pages (is_platform_page=False, vendor_id=NULL) + - Fallback pages for vendors who haven't customized + - About Us, Shipping Policy, Return Policy, etc. + +3. Vendor Override/Custom Pages (is_platform_page=False, vendor_id=set) + - Vendor-specific customizations + - Either overrides a default or is a completely custom page + +Lookup Strategy for Vendor Storefronts: +1. Check for vendor override (platform_id + vendor_id + slug + published) +2. If not found, check for vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) +3. If neither exists, return None/404 +""" + +import logging +from datetime import UTC, datetime + +from sqlalchemy import and_ +from sqlalchemy.orm import Session + +from app.modules.cms.exceptions import ( + ContentPageNotFoundException, + UnauthorizedContentPageAccessException, +) +# Import from canonical location to avoid circular imports +from models.database.content_page import ContentPage + +logger = logging.getLogger(__name__) + + +class ContentPageService: + """Service for content page operations with multi-platform support.""" + + # ========================================================================= + # Three-Tier Resolution Methods (for vendor storefronts) + # ========================================================================= + + @staticmethod + def get_page_for_vendor( + db: Session, + platform_id: int, + slug: str, + vendor_id: int | None = None, + include_unpublished: bool = False, + ) -> ContentPage | None: + """ + Get content page with three-tier resolution. + + Resolution order: + 1. Vendor override (platform_id + vendor_id + slug) + 2. Vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) + + Args: + db: Database session + platform_id: Platform ID (required for multi-platform support) + slug: Page slug (about, faq, contact, etc.) + vendor_id: Vendor ID (None for defaults only) + include_unpublished: Include draft pages (for preview) + + Returns: + ContentPage or None + """ + base_filters = [ + ContentPage.platform_id == platform_id, + ContentPage.slug == slug, + ] + + if not include_unpublished: + base_filters.append(ContentPage.is_published == True) + + # Tier 1: Try vendor-specific override first + if vendor_id: + vendor_page = ( + db.query(ContentPage) + .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) + .first() + ) + + if vendor_page: + logger.debug( + f"[CMS] Found vendor override: {slug} for vendor_id={vendor_id}, platform_id={platform_id}" + ) + return vendor_page + + # Tier 2: Fallback to vendor default (not platform page) + vendor_default_page = ( + db.query(ContentPage) + .filter( + and_( + ContentPage.vendor_id == None, + ContentPage.is_platform_page == False, + *base_filters, + ) + ) + .first() + ) + + if vendor_default_page: + logger.debug(f"[CMS] Using vendor default page: {slug} for platform_id={platform_id}") + return vendor_default_page + + logger.debug(f"[CMS] No page found for slug: {slug}, platform_id={platform_id}") + return None + + @staticmethod + def get_platform_page( + db: Session, + platform_id: int, + slug: str, + include_unpublished: bool = False, + ) -> ContentPage | None: + """ + Get a platform marketing page. + + Platform marketing pages are pages that describe the platform itself + (homepage, pricing, about, features, etc.). + + Args: + db: Database session + platform_id: Platform ID + slug: Page slug + include_unpublished: Include draft pages + + Returns: + ContentPage or None + """ + filters = [ + ContentPage.platform_id == platform_id, + ContentPage.slug == slug, + ContentPage.vendor_id == None, + ContentPage.is_platform_page == True, + ] + + if not include_unpublished: + filters.append(ContentPage.is_published == True) + + page = db.query(ContentPage).filter(and_(*filters)).first() + + if page: + logger.debug(f"[CMS] Found platform page: {slug} for platform_id={platform_id}") + else: + logger.debug(f"[CMS] No platform page found: {slug} for platform_id={platform_id}") + + return page + + @staticmethod + def list_pages_for_vendor( + db: Session, + platform_id: int, + vendor_id: int | None = None, + include_unpublished: bool = False, + footer_only: bool = False, + header_only: bool = False, + legal_only: bool = False, + ) -> list[ContentPage]: + """ + List all available pages for a vendor storefront. + + Merges vendor overrides with vendor defaults, prioritizing overrides. + Does NOT include platform marketing pages. + + Args: + db: Database session + platform_id: Platform ID + vendor_id: Vendor ID (None for vendor defaults only) + include_unpublished: Include draft pages + footer_only: Only pages marked for footer display + header_only: Only pages marked for header display + legal_only: Only pages marked for legal/bottom bar display + + Returns: + List of ContentPage objects + """ + base_filters = [ContentPage.platform_id == platform_id] + + if not include_unpublished: + base_filters.append(ContentPage.is_published == True) + + if footer_only: + base_filters.append(ContentPage.show_in_footer == True) + + if header_only: + base_filters.append(ContentPage.show_in_header == True) + + if legal_only: + base_filters.append(ContentPage.show_in_legal == True) + + # Get vendor-specific pages + vendor_pages = [] + if vendor_id: + vendor_pages = ( + db.query(ContentPage) + .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) + .order_by(ContentPage.display_order, ContentPage.title) + .all() + ) + + # Get vendor defaults (not platform marketing pages) + vendor_default_pages = ( + db.query(ContentPage) + .filter( + and_( + ContentPage.vendor_id == None, + ContentPage.is_platform_page == False, + *base_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 vendor_default_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 list_platform_pages( + db: Session, + platform_id: int, + include_unpublished: bool = False, + footer_only: bool = False, + header_only: bool = False, + ) -> list[ContentPage]: + """ + List platform marketing pages. + + Args: + db: Database session + platform_id: Platform ID + include_unpublished: Include draft pages + footer_only: Only pages marked for footer display + header_only: Only pages marked for header display + + Returns: + List of platform marketing ContentPage objects + """ + filters = [ + ContentPage.platform_id == platform_id, + ContentPage.vendor_id == None, + ContentPage.is_platform_page == True, + ] + + 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) + + return ( + db.query(ContentPage) + .filter(and_(*filters)) + .order_by(ContentPage.display_order, ContentPage.title) + .all() + ) + + @staticmethod + def list_vendor_defaults( + db: Session, + platform_id: int, + include_unpublished: bool = False, + ) -> list[ContentPage]: + """ + List vendor default pages (fallbacks for vendors who haven't customized). + + Args: + db: Database session + platform_id: Platform ID + include_unpublished: Include draft pages + + Returns: + List of vendor default ContentPage objects + """ + filters = [ + ContentPage.platform_id == platform_id, + ContentPage.vendor_id == None, + ContentPage.is_platform_page == False, + ] + + 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 all platform marketing pages across all platforms (for admin use). + + Args: + db: Database session + include_unpublished: Include draft pages + + Returns: + List of all platform marketing ContentPage objects + """ + filters = [ + ContentPage.vendor_id.is_(None), + ContentPage.is_platform_page.is_(True), + ] + + if not include_unpublished: + filters.append(ContentPage.is_published.is_(True)) + + return ( + db.query(ContentPage) + .filter(and_(*filters)) + .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) + .all() + ) + + @staticmethod + def list_all_vendor_defaults( + db: Session, + include_unpublished: bool = False, + ) -> list[ContentPage]: + """ + List all vendor default pages across all platforms (for admin use). + + Args: + db: Database session + include_unpublished: Include draft pages + + Returns: + List of all vendor default ContentPage objects + """ + filters = [ + ContentPage.vendor_id.is_(None), + ContentPage.is_platform_page.is_(False), + ] + + if not include_unpublished: + filters.append(ContentPage.is_published.is_(True)) + + return ( + db.query(ContentPage) + .filter(and_(*filters)) + .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) + .all() + ) + + # ========================================================================= + # CRUD Methods + # ========================================================================= + + @staticmethod + def create_page( + db: Session, + platform_id: int, + slug: str, + title: str, + content: str, + vendor_id: int | None = None, + is_platform_page: bool = False, + content_format: str = "html", + template: str = "default", + meta_description: str | None = None, + meta_keywords: str | None = None, + is_published: bool = False, + show_in_footer: bool = True, + show_in_header: bool = False, + show_in_legal: bool = False, + display_order: int = 0, + created_by: int | None = None, + ) -> ContentPage: + """ + Create a new content page. + + Args: + db: Database session + platform_id: Platform ID (required) + slug: URL-safe identifier + title: Page title + content: HTML or Markdown content + vendor_id: Vendor ID (None for platform/default pages) + is_platform_page: True for platform marketing pages + content_format: "html" or "markdown" + template: Template name for landing pages + 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 + show_in_legal: Show in legal/bottom bar navigation + display_order: Sort order + created_by: User ID who created it + + Returns: + Created ContentPage + """ + page = ContentPage( + platform_id=platform_id, + vendor_id=vendor_id, + is_platform_page=is_platform_page, + slug=slug, + title=title, + content=content, + content_format=content_format, + template=template, + meta_description=meta_description, + meta_keywords=meta_keywords, + is_published=is_published, + published_at=datetime.now(UTC) if is_published else None, + show_in_footer=show_in_footer, + show_in_header=show_in_header, + show_in_legal=show_in_legal, + display_order=display_order, + created_by=created_by, + updated_by=created_by, + ) + + db.add(page) + db.flush() + db.refresh(page) + + page_type = "platform" if is_platform_page else ("vendor" if vendor_id else "default") + logger.info( + f"[CMS] Created {page_type} page: {slug} (platform_id={platform_id}, vendor_id={vendor_id}, id={page.id})" + ) + return page + + @staticmethod + def update_page( + db: Session, + page_id: int, + title: str | None = None, + content: str | None = None, + content_format: str | None = None, + template: str | None = None, + meta_description: str | None = None, + meta_keywords: str | None = None, + is_published: bool | None = None, + show_in_footer: bool | None = None, + show_in_header: bool | None = None, + show_in_legal: bool | None = None, + display_order: int | None = None, + updated_by: int | None = None, + ) -> ContentPage | None: + """ + Update an existing content page. + + Args: + db: Database session + page_id: Page ID + ... other fields + + Returns: + Updated ContentPage or None if not found + """ + page = db.query(ContentPage).filter(ContentPage.id == page_id).first() + + if not page: + logger.warning(f"[CMS] 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 template is not None: + page.template = template + 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(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 show_in_legal is not None: + page.show_in_legal = show_in_legal + if display_order is not None: + page.display_order = display_order + if updated_by is not None: + page.updated_by = updated_by + + db.flush() + db.refresh(page) + + logger.info(f"[CMS] 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"[CMS] Content page not found for deletion: id={page_id}") + return False + + db.delete(page) + + logger.info(f"[CMS] Deleted content page: id={page_id}, slug={page.slug}") + return True + + @staticmethod + def get_page_by_id(db: Session, page_id: int) -> ContentPage | None: + """Get content page by ID.""" + return db.query(ContentPage).filter(ContentPage.id == page_id).first() + + @staticmethod + def get_page_by_id_or_raise(db: Session, page_id: int) -> ContentPage: + """ + Get content page by ID or raise ContentPageNotFoundException. + """ + page = db.query(ContentPage).filter(ContentPage.id == page_id).first() + if not page: + raise ContentPageNotFoundException(identifier=page_id) + return page + + @staticmethod + def get_page_for_vendor_or_raise( + db: Session, + platform_id: int, + slug: str, + vendor_id: int | None = None, + include_unpublished: bool = False, + ) -> ContentPage: + """ + Get content page for a vendor with three-tier resolution. + Raises ContentPageNotFoundException if not found. + """ + page = ContentPageService.get_page_for_vendor( + db, + platform_id=platform_id, + slug=slug, + vendor_id=vendor_id, + include_unpublished=include_unpublished, + ) + if not page: + raise ContentPageNotFoundException(identifier=slug) + return page + + @staticmethod + def get_platform_page_or_raise( + db: Session, + platform_id: int, + slug: str, + include_unpublished: bool = False, + ) -> ContentPage: + """ + Get platform marketing page or raise ContentPageNotFoundException. + """ + page = ContentPageService.get_platform_page( + db, + platform_id=platform_id, + slug=slug, + include_unpublished=include_unpublished, + ) + if not page: + raise ContentPageNotFoundException(identifier=slug) + return page + + # ========================================================================= + # Vendor Page Management (with ownership checks) + # ========================================================================= + + @staticmethod + def update_vendor_page( + db: Session, + page_id: int, + vendor_id: int, + title: str | None = None, + content: str | None = None, + content_format: str | None = None, + meta_description: str | None = None, + meta_keywords: str | None = None, + is_published: bool | None = None, + show_in_footer: bool | None = None, + show_in_header: bool | None = None, + show_in_legal: bool | None = None, + display_order: int | None = None, + updated_by: int | None = None, + ) -> ContentPage: + """ + Update a vendor-specific content page with ownership check. + + Raises: + ContentPageNotFoundException: If page not found + UnauthorizedContentPageAccessException: If page doesn't belong to vendor + """ + page = ContentPageService.get_page_by_id_or_raise(db, page_id) + + if page.vendor_id != vendor_id: + raise UnauthorizedContentPageAccessException(action="edit") + + return ContentPageService.update_page( + db, + page_id=page_id, + title=title, + content=content, + content_format=content_format, + meta_description=meta_description, + meta_keywords=meta_keywords, + is_published=is_published, + show_in_footer=show_in_footer, + show_in_header=show_in_header, + show_in_legal=show_in_legal, + display_order=display_order, + updated_by=updated_by, + ) + + @staticmethod + def delete_vendor_page(db: Session, page_id: int, vendor_id: int) -> None: + """ + Delete a vendor-specific content page with ownership check. + + Raises: + ContentPageNotFoundException: If page not found + UnauthorizedContentPageAccessException: If page doesn't belong to vendor + """ + page = ContentPageService.get_page_by_id_or_raise(db, page_id) + + if page.vendor_id != vendor_id: + raise UnauthorizedContentPageAccessException(action="delete") + + ContentPageService.delete_page(db, page_id) + + @staticmethod + def create_vendor_override( + db: Session, + platform_id: int, + vendor_id: int, + slug: str, + title: str, + content: str, + content_format: str = "html", + meta_description: str | None = None, + meta_keywords: str | None = None, + is_published: bool = False, + show_in_footer: bool = True, + show_in_header: bool = False, + show_in_legal: bool = False, + display_order: int = 0, + created_by: int | None = None, + ) -> ContentPage: + """ + Create a vendor override page (vendor-specific customization of a default). + + Args: + db: Database session + platform_id: Platform ID + vendor_id: Vendor ID + slug: Page slug (typically matches a default page) + ... other fields + + Returns: + Created ContentPage + """ + return ContentPageService.create_page( + db, + platform_id=platform_id, + slug=slug, + title=title, + content=content, + vendor_id=vendor_id, + is_platform_page=False, + content_format=content_format, + meta_description=meta_description, + meta_keywords=meta_keywords, + is_published=is_published, + show_in_footer=show_in_footer, + show_in_header=show_in_header, + show_in_legal=show_in_legal, + display_order=display_order, + created_by=created_by, + ) + + @staticmethod + def revert_to_default(db: Session, page_id: int, vendor_id: int) -> None: + """ + Revert a vendor override to the default by deleting the override. + + After deletion, the vendor storefront will use the vendor default page. + + Raises: + ContentPageNotFoundException: If page not found + UnauthorizedContentPageAccessException: If page doesn't belong to vendor + """ + ContentPageService.delete_vendor_page(db, page_id, vendor_id) + + # ========================================================================= + # Admin Methods (for listing all pages) + # ========================================================================= + + @staticmethod + def list_all_pages( + db: Session, + platform_id: int | None = None, + vendor_id: int | None = None, + include_unpublished: bool = False, + page_tier: str | None = None, + ) -> list[ContentPage]: + """ + List all content pages for admin management. + + Args: + db: Database session + platform_id: Optional filter by platform ID + vendor_id: Optional filter by vendor ID + include_unpublished: Include draft pages + page_tier: Optional filter by tier ("platform", "vendor_default", "vendor_override") + + Returns: + List of ContentPage objects + """ + filters = [] + + if platform_id: + filters.append(ContentPage.platform_id == platform_id) + + if vendor_id is not None: + filters.append(ContentPage.vendor_id == vendor_id) + + if not include_unpublished: + filters.append(ContentPage.is_published == True) + + if page_tier == "platform": + filters.append(ContentPage.is_platform_page == True) + filters.append(ContentPage.vendor_id == None) + elif page_tier == "vendor_default": + filters.append(ContentPage.is_platform_page == False) + filters.append(ContentPage.vendor_id == None) + elif page_tier == "vendor_override": + filters.append(ContentPage.vendor_id != None) + + return ( + db.query(ContentPage) + .filter(and_(*filters) if filters else True) + .order_by( + ContentPage.platform_id, + ContentPage.vendor_id, + ContentPage.display_order, + ContentPage.title, + ) + .all() + ) + + @staticmethod + def list_all_vendor_pages( + db: Session, + vendor_id: int, + platform_id: int | None = None, + include_unpublished: bool = False, + ) -> list[ContentPage]: + """ + List only vendor-specific pages (overrides and custom pages). + + Args: + db: Database session + vendor_id: Vendor ID + platform_id: Optional filter by platform + include_unpublished: Include draft pages + + Returns: + List of vendor-specific ContentPage objects + """ + filters = [ContentPage.vendor_id == vendor_id] + + if platform_id: + filters.append(ContentPage.platform_id == platform_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() + ) + + # ========================================================================= + # Helper Methods for raising exceptions + # ========================================================================= + + @staticmethod + def update_page_or_raise( + db: Session, + page_id: int, + title: str | None = None, + content: str | None = None, + content_format: str | None = None, + template: str | None = None, + meta_description: str | None = None, + meta_keywords: str | None = None, + is_published: bool | None = None, + show_in_footer: bool | None = None, + show_in_header: bool | None = None, + show_in_legal: bool | None = None, + display_order: int | None = None, + updated_by: int | None = None, + ) -> ContentPage: + """ + Update an existing content page or raise exception. + + Raises: + ContentPageNotFoundException: If page not found + """ + page = ContentPageService.update_page( + db, + page_id=page_id, + title=title, + content=content, + content_format=content_format, + template=template, + meta_description=meta_description, + meta_keywords=meta_keywords, + is_published=is_published, + show_in_footer=show_in_footer, + show_in_header=show_in_header, + show_in_legal=show_in_legal, + display_order=display_order, + updated_by=updated_by, + ) + if not page: + raise ContentPageNotFoundException(identifier=page_id) + return page + + @staticmethod + def delete_page_or_raise(db: Session, page_id: int) -> None: + """ + Delete a content page or raise exception. + + Raises: + ContentPageNotFoundException: If page not found + """ + success = ContentPageService.delete_page(db, page_id) + if not success: + raise ContentPageNotFoundException(identifier=page_id) + + # ========================================================================= + # Homepage Sections Management + # ========================================================================= + + @staticmethod + def update_homepage_sections( + db: Session, + page_id: int, + sections: dict, + updated_by: int | None = None, + ) -> ContentPage: + """ + Update homepage sections with validation. + + Args: + db: Database session + page_id: Content page ID + sections: Homepage sections dict (validated against HomepageSections schema) + updated_by: User ID making the update + + Returns: + Updated ContentPage + + Raises: + ContentPageNotFoundException: If page not found + ValidationError: If sections schema invalid + """ + from models.schema.homepage_sections import HomepageSections + + page = ContentPageService.get_page_by_id_or_raise(db, page_id) + + # Validate sections against schema + validated = HomepageSections(**sections) + + # Update page + page.sections = validated.model_dump() + page.updated_by = updated_by + + db.flush() + db.refresh(page) + + logger.info(f"[CMS] Updated homepage sections for page_id={page_id}") + return page + + @staticmethod + def update_single_section( + db: Session, + page_id: int, + section_name: str, + section_data: dict, + updated_by: int | None = None, + ) -> ContentPage: + """ + Update a single section within homepage sections. + + Args: + db: Database session + page_id: Content page ID + section_name: Section to update (hero, features, pricing, cta) + section_data: Section configuration dict + updated_by: User ID making the update + + Returns: + Updated ContentPage + + Raises: + ContentPageNotFoundException: If page not found + ValueError: If section name is invalid + """ + from models.schema.homepage_sections import ( + HeroSection, + FeaturesSection, + PricingSection, + CTASection, + ) + + SECTION_SCHEMAS = { + "hero": HeroSection, + "features": FeaturesSection, + "pricing": PricingSection, + "cta": CTASection, + } + + if section_name not in SECTION_SCHEMAS: + raise ValueError(f"Invalid section name: {section_name}. Must be one of: {list(SECTION_SCHEMAS.keys())}") + + page = ContentPageService.get_page_by_id_or_raise(db, page_id) + + # Validate section data against its schema + schema = SECTION_SCHEMAS[section_name] + validated_section = schema(**section_data) + + # Initialize sections if needed + current_sections = page.sections or {} + current_sections[section_name] = validated_section.model_dump() + + page.sections = current_sections + page.updated_by = updated_by + + db.flush() + db.refresh(page) + + logger.info(f"[CMS] Updated section '{section_name}' for page_id={page_id}") + return page + + @staticmethod + def get_default_sections(languages: list[str] | None = None) -> dict: + """ + Get empty sections structure for new homepage. + + Args: + languages: List of language codes from platform.supported_languages. + Defaults to ['fr', 'de', 'en'] if not provided. + + Returns: + Empty sections dict with language placeholders + """ + from models.schema.homepage_sections import HomepageSections + + if languages is None: + languages = ["fr", "de", "en"] + + return HomepageSections.get_empty_structure(languages).model_dump() + + +# Singleton instance +content_page_service = ContentPageService() diff --git a/app/modules/contracts/__init__.py b/app/modules/contracts/__init__.py new file mode 100644 index 00000000..b3d3faf5 --- /dev/null +++ b/app/modules/contracts/__init__.py @@ -0,0 +1,32 @@ +# app/modules/contracts/__init__.py +""" +Cross-module contracts using Protocol pattern. + +This module defines type-safe interfaces for cross-module communication. +Modules depend on protocols rather than concrete implementations, enabling: +- Loose coupling between modules +- Testability through mock implementations +- Clear dependency boundaries + +Usage: + from app.modules.contracts.cms import ContentServiceProtocol + + class OrderService: + def __init__(self, content: ContentServiceProtocol | None = None): + self._content = content + + @property + def content(self) -> ContentServiceProtocol: + if self._content is None: + from app.modules.cms.services import content_page_service + self._content = content_page_service + return self._content +""" + +from app.modules.contracts.base import ServiceProtocol +from app.modules.contracts.cms import ContentServiceProtocol + +__all__ = [ + "ServiceProtocol", + "ContentServiceProtocol", +] diff --git a/app/modules/contracts/base.py b/app/modules/contracts/base.py new file mode 100644 index 00000000..17a7193c --- /dev/null +++ b/app/modules/contracts/base.py @@ -0,0 +1,47 @@ +# app/modules/contracts/base.py +""" +Base protocol definitions for cross-module communication. + +Protocols define the interface that services must implement without +requiring inheritance. This allows for: +- Duck typing with static type checking +- Easy mocking in tests +- Module independence +""" + +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + + +@runtime_checkable +class ServiceProtocol(Protocol): + """ + Base protocol for all module services. + + Services should be stateless and operate on database sessions + passed as arguments. This ensures: + - Thread safety + - Transaction boundaries are clear + - Easy testing with mock sessions + """ + + pass + + +@runtime_checkable +class CRUDServiceProtocol(Protocol): + """ + Protocol for services that provide CRUD operations. + + All methods receive a database session as the first argument. + """ + + def get_by_id(self, db: "Session", id: int) -> object | None: + """Get entity by ID.""" + ... + + def delete(self, db: "Session", id: int) -> bool: + """Delete entity by ID. Returns True if deleted.""" + ... diff --git a/app/modules/contracts/cms.py b/app/modules/contracts/cms.py new file mode 100644 index 00000000..ea17e9c1 --- /dev/null +++ b/app/modules/contracts/cms.py @@ -0,0 +1,148 @@ +# app/modules/contracts/cms.py +""" +CMS module contracts. + +Defines the interface for content management functionality that other +modules can depend on without importing the concrete implementation. +""" + +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + + +@runtime_checkable +class ContentServiceProtocol(Protocol): + """ + Protocol for content page service. + + Defines the interface for retrieving and managing content pages + with three-tier resolution (platform > vendor default > vendor override). + """ + + def get_page_for_vendor( + self, + db: "Session", + platform_id: int, + slug: str, + vendor_id: int | None = None, + include_unpublished: bool = False, + ) -> object | None: + """ + Get content page with three-tier resolution. + + Resolution order: + 1. Vendor override (platform_id + vendor_id + slug) + 2. Vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) + + Args: + db: Database session + platform_id: Platform ID + slug: Page slug + vendor_id: Vendor ID (None for defaults only) + include_unpublished: Include draft pages + + Returns: + ContentPage or None + """ + ... + + def get_platform_page( + self, + db: "Session", + platform_id: int, + slug: str, + include_unpublished: bool = False, + ) -> object | None: + """ + Get a platform marketing page. + + Args: + db: Database session + platform_id: Platform ID + slug: Page slug + include_unpublished: Include draft pages + + Returns: + ContentPage or None + """ + ... + + def list_pages_for_vendor( + self, + db: "Session", + platform_id: int, + vendor_id: int | None = None, + include_unpublished: bool = False, + footer_only: bool = False, + header_only: bool = False, + legal_only: bool = False, + ) -> list: + """ + List all available pages for a vendor storefront. + + Merges vendor overrides with vendor defaults, prioritizing overrides. + + Args: + db: Database session + platform_id: Platform ID + vendor_id: Vendor ID (None for vendor defaults only) + include_unpublished: Include draft pages + footer_only: Only pages marked for footer display + header_only: Only pages marked for header display + legal_only: Only pages marked for legal/bottom bar display + + Returns: + List of ContentPage objects + """ + ... + + def list_platform_pages( + self, + db: "Session", + platform_id: int, + include_unpublished: bool = False, + footer_only: bool = False, + header_only: bool = False, + ) -> list: + """ + List platform marketing pages. + + Args: + db: Database session + platform_id: Platform ID + include_unpublished: Include draft pages + footer_only: Only pages marked for footer display + header_only: Only pages marked for header display + + Returns: + List of platform marketing ContentPage objects + """ + ... + + +@runtime_checkable +class MediaServiceProtocol(Protocol): + """ + Protocol for media library service. + + Defines the interface for managing media files (images, documents, etc.). + """ + + def get_media_by_id( + self, + db: "Session", + media_id: int, + ) -> object | None: + """Get media item by ID.""" + ... + + def list_media_for_vendor( + self, + db: "Session", + vendor_id: int, + media_type: str | None = None, + ) -> list: + """List media items for a vendor.""" + ... diff --git a/app/services/content_page_service.py b/app/services/content_page_service.py index b930e2ab..e79ae9e8 100644 --- a/app/services/content_page_service.py +++ b/app/services/content_page_service.py @@ -1,1001 +1,30 @@ # app/services/content_page_service.py """ -Content Page Service +DEPRECATED: This module has moved to app.modules.cms.services.content_page_service -Business logic for managing content pages with three-tier hierarchy: +Please update your imports: + # Old (deprecated): + from app.services.content_page_service import content_page_service -1. Platform Marketing Pages (is_platform_page=True, vendor_id=NULL) - - Platform's own pages (homepage, pricing, about) - - Describe the platform/business offering itself + # New (preferred): + from app.modules.cms.services import content_page_service -2. Vendor Default Pages (is_platform_page=False, vendor_id=NULL) - - Fallback pages for vendors who haven't customized - - About Us, Shipping Policy, Return Policy, etc. - -3. Vendor Override/Custom Pages (is_platform_page=False, vendor_id=set) - - Vendor-specific customizations - - Either overrides a default or is a completely custom page - -Lookup Strategy for Vendor Storefronts: -1. Check for vendor override (platform_id + vendor_id + slug + published) -2. If not found, check for vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) -3. If neither exists, return None/404 +This shim re-exports from the new location for backwards compatibility. """ -import logging -from datetime import UTC, datetime +import warnings -from sqlalchemy import and_ -from sqlalchemy.orm import Session - -from app.exceptions.content_page import ( - ContentPageNotFoundException, - UnauthorizedContentPageAccessException, +warnings.warn( + "Import from app.modules.cms.services.content_page_service instead of " + "app.services.content_page_service. This shim will be removed in a future version.", + DeprecationWarning, + stacklevel=2, ) -from models.database.content_page import ContentPage -logger = logging.getLogger(__name__) +# Re-export everything from the new location +from app.modules.cms.services.content_page_service import ( # noqa: E402, F401 + ContentPageService, + content_page_service, +) - -class ContentPageService: - """Service for content page operations with multi-platform support.""" - - # ========================================================================= - # Three-Tier Resolution Methods (for vendor storefronts) - # ========================================================================= - - @staticmethod - def get_page_for_vendor( - db: Session, - platform_id: int, - slug: str, - vendor_id: int | None = None, - include_unpublished: bool = False, - ) -> ContentPage | None: - """ - Get content page for a vendor with three-tier resolution. - - Resolution order: - 1. Vendor override (platform_id + vendor_id + slug) - 2. Vendor default (platform_id + vendor_id=NULL + is_platform_page=False + slug) - - Args: - db: Database session - platform_id: Platform ID (required for multi-platform support) - slug: Page slug (about, faq, contact, etc.) - vendor_id: Vendor ID (None for defaults only) - include_unpublished: Include draft pages (for preview) - - Returns: - ContentPage or None - """ - base_filters = [ - ContentPage.platform_id == platform_id, - ContentPage.slug == slug, - ] - - if not include_unpublished: - base_filters.append(ContentPage.is_published == True) - - # Tier 1: Try vendor-specific override first - if vendor_id: - vendor_page = ( - db.query(ContentPage) - .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) - .first() - ) - - if vendor_page: - logger.debug( - f"[CMS] Found vendor override: {slug} for vendor_id={vendor_id}, platform_id={platform_id}" - ) - return vendor_page - - # Tier 2: Fallback to vendor default (not platform page) - vendor_default_page = ( - db.query(ContentPage) - .filter( - and_( - ContentPage.vendor_id == None, - ContentPage.is_platform_page == False, - *base_filters, - ) - ) - .first() - ) - - if vendor_default_page: - logger.debug(f"[CMS] Using vendor default page: {slug} for platform_id={platform_id}") - return vendor_default_page - - logger.debug(f"[CMS] No page found for slug: {slug}, platform_id={platform_id}") - return None - - @staticmethod - def get_platform_page( - db: Session, - platform_id: int, - slug: str, - include_unpublished: bool = False, - ) -> ContentPage | None: - """ - Get a platform marketing page. - - Platform marketing pages are pages that describe the platform itself - (homepage, pricing, about, features, etc.). - - Args: - db: Database session - platform_id: Platform ID - slug: Page slug - include_unpublished: Include draft pages - - Returns: - ContentPage or None - """ - filters = [ - ContentPage.platform_id == platform_id, - ContentPage.slug == slug, - ContentPage.vendor_id == None, - ContentPage.is_platform_page == True, - ] - - if not include_unpublished: - filters.append(ContentPage.is_published == True) - - page = db.query(ContentPage).filter(and_(*filters)).first() - - if page: - logger.debug(f"[CMS] Found platform page: {slug} for platform_id={platform_id}") - else: - logger.debug(f"[CMS] No platform page found: {slug} for platform_id={platform_id}") - - return page - - @staticmethod - def list_pages_for_vendor( - db: Session, - platform_id: int, - vendor_id: int | None = None, - include_unpublished: bool = False, - footer_only: bool = False, - header_only: bool = False, - legal_only: bool = False, - ) -> list[ContentPage]: - """ - List all available pages for a vendor storefront. - - Merges vendor overrides with vendor defaults, prioritizing overrides. - Does NOT include platform marketing pages. - - Args: - db: Database session - platform_id: Platform ID - vendor_id: Vendor ID (None for vendor defaults only) - include_unpublished: Include draft pages - footer_only: Only pages marked for footer display - header_only: Only pages marked for header display - legal_only: Only pages marked for legal/bottom bar display - - Returns: - List of ContentPage objects - """ - base_filters = [ContentPage.platform_id == platform_id] - - if not include_unpublished: - base_filters.append(ContentPage.is_published == True) - - if footer_only: - base_filters.append(ContentPage.show_in_footer == True) - - if header_only: - base_filters.append(ContentPage.show_in_header == True) - - if legal_only: - base_filters.append(ContentPage.show_in_legal == True) - - # Get vendor-specific pages - vendor_pages = [] - if vendor_id: - vendor_pages = ( - db.query(ContentPage) - .filter(and_(ContentPage.vendor_id == vendor_id, *base_filters)) - .order_by(ContentPage.display_order, ContentPage.title) - .all() - ) - - # Get vendor defaults (not platform marketing pages) - vendor_default_pages = ( - db.query(ContentPage) - .filter( - and_( - ContentPage.vendor_id == None, - ContentPage.is_platform_page == False, - *base_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 vendor_default_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 list_platform_pages( - db: Session, - platform_id: int, - include_unpublished: bool = False, - footer_only: bool = False, - header_only: bool = False, - ) -> list[ContentPage]: - """ - List platform marketing pages. - - Args: - db: Database session - platform_id: Platform ID - include_unpublished: Include draft pages - footer_only: Only pages marked for footer display - header_only: Only pages marked for header display - - Returns: - List of platform marketing ContentPage objects - """ - filters = [ - ContentPage.platform_id == platform_id, - ContentPage.vendor_id == None, - ContentPage.is_platform_page == True, - ] - - 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) - - return ( - db.query(ContentPage) - .filter(and_(*filters)) - .order_by(ContentPage.display_order, ContentPage.title) - .all() - ) - - @staticmethod - def list_vendor_defaults( - db: Session, - platform_id: int, - include_unpublished: bool = False, - ) -> list[ContentPage]: - """ - List vendor default pages (fallbacks for vendors who haven't customized). - - Args: - db: Database session - platform_id: Platform ID - include_unpublished: Include draft pages - - Returns: - List of vendor default ContentPage objects - """ - filters = [ - ContentPage.platform_id == platform_id, - ContentPage.vendor_id == None, - ContentPage.is_platform_page == False, - ] - - 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 all platform marketing pages across all platforms (for admin use). - - Args: - db: Database session - include_unpublished: Include draft pages - - Returns: - List of all platform marketing ContentPage objects - """ - filters = [ - ContentPage.vendor_id.is_(None), - ContentPage.is_platform_page.is_(True), - ] - - if not include_unpublished: - filters.append(ContentPage.is_published.is_(True)) - - return ( - db.query(ContentPage) - .filter(and_(*filters)) - .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) - .all() - ) - - @staticmethod - def list_all_vendor_defaults( - db: Session, - include_unpublished: bool = False, - ) -> list[ContentPage]: - """ - List all vendor default pages across all platforms (for admin use). - - Args: - db: Database session - include_unpublished: Include draft pages - - Returns: - List of all vendor default ContentPage objects - """ - filters = [ - ContentPage.vendor_id.is_(None), - ContentPage.is_platform_page.is_(False), - ] - - if not include_unpublished: - filters.append(ContentPage.is_published.is_(True)) - - return ( - db.query(ContentPage) - .filter(and_(*filters)) - .order_by(ContentPage.platform_id, ContentPage.display_order, ContentPage.title) - .all() - ) - - # ========================================================================= - # CRUD Methods - # ========================================================================= - - @staticmethod - def create_page( - db: Session, - platform_id: int, - slug: str, - title: str, - content: str, - vendor_id: int | None = None, - is_platform_page: bool = False, - content_format: str = "html", - template: str = "default", - meta_description: str | None = None, - meta_keywords: str | None = None, - is_published: bool = False, - show_in_footer: bool = True, - show_in_header: bool = False, - show_in_legal: bool = False, - display_order: int = 0, - created_by: int | None = None, - ) -> ContentPage: - """ - Create a new content page. - - Args: - db: Database session - platform_id: Platform ID (required) - slug: URL-safe identifier - title: Page title - content: HTML or Markdown content - vendor_id: Vendor ID (None for platform/default pages) - is_platform_page: True for platform marketing pages - content_format: "html" or "markdown" - template: Template name for landing pages - 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 - show_in_legal: Show in legal/bottom bar navigation - display_order: Sort order - created_by: User ID who created it - - Returns: - Created ContentPage - """ - page = ContentPage( - platform_id=platform_id, - vendor_id=vendor_id, - is_platform_page=is_platform_page, - slug=slug, - title=title, - content=content, - content_format=content_format, - template=template, - meta_description=meta_description, - meta_keywords=meta_keywords, - is_published=is_published, - published_at=datetime.now(UTC) if is_published else None, - show_in_footer=show_in_footer, - show_in_header=show_in_header, - show_in_legal=show_in_legal, - display_order=display_order, - created_by=created_by, - updated_by=created_by, - ) - - db.add(page) - db.flush() - db.refresh(page) - - page_type = "platform" if is_platform_page else ("vendor" if vendor_id else "default") - logger.info( - f"[CMS] Created {page_type} page: {slug} (platform_id={platform_id}, vendor_id={vendor_id}, id={page.id})" - ) - return page - - @staticmethod - def update_page( - db: Session, - page_id: int, - title: str | None = None, - content: str | None = None, - content_format: str | None = None, - template: str | None = None, - meta_description: str | None = None, - meta_keywords: str | None = None, - is_published: bool | None = None, - show_in_footer: bool | None = None, - show_in_header: bool | None = None, - show_in_legal: bool | None = None, - display_order: int | None = None, - updated_by: int | None = None, - ) -> ContentPage | None: - """ - Update an existing content page. - - Args: - db: Database session - page_id: Page ID - ... other fields - - Returns: - Updated ContentPage or None if not found - """ - page = db.query(ContentPage).filter(ContentPage.id == page_id).first() - - if not page: - logger.warning(f"[CMS] 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 template is not None: - page.template = template - 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(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 show_in_legal is not None: - page.show_in_legal = show_in_legal - if display_order is not None: - page.display_order = display_order - if updated_by is not None: - page.updated_by = updated_by - - db.flush() - db.refresh(page) - - logger.info(f"[CMS] 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"[CMS] Content page not found for deletion: id={page_id}") - return False - - db.delete(page) - - logger.info(f"[CMS] Deleted content page: id={page_id}, slug={page.slug}") - return True - - @staticmethod - def get_page_by_id(db: Session, page_id: int) -> ContentPage | None: - """Get content page by ID.""" - return db.query(ContentPage).filter(ContentPage.id == page_id).first() - - @staticmethod - def get_page_by_id_or_raise(db: Session, page_id: int) -> ContentPage: - """ - Get content page by ID or raise ContentPageNotFoundException. - """ - page = db.query(ContentPage).filter(ContentPage.id == page_id).first() - if not page: - raise ContentPageNotFoundException(identifier=page_id) - return page - - @staticmethod - def get_page_for_vendor_or_raise( - db: Session, - platform_id: int, - slug: str, - vendor_id: int | None = None, - include_unpublished: bool = False, - ) -> ContentPage: - """ - Get content page for a vendor with three-tier resolution. - Raises ContentPageNotFoundException if not found. - """ - page = ContentPageService.get_page_for_vendor( - db, - platform_id=platform_id, - slug=slug, - vendor_id=vendor_id, - include_unpublished=include_unpublished, - ) - if not page: - raise ContentPageNotFoundException(identifier=slug) - return page - - @staticmethod - def get_platform_page_or_raise( - db: Session, - platform_id: int, - slug: str, - include_unpublished: bool = False, - ) -> ContentPage: - """ - Get platform marketing page or raise ContentPageNotFoundException. - """ - page = ContentPageService.get_platform_page( - db, - platform_id=platform_id, - slug=slug, - include_unpublished=include_unpublished, - ) - if not page: - raise ContentPageNotFoundException(identifier=slug) - return page - - # ========================================================================= - # Vendor Page Management (with ownership checks) - # ========================================================================= - - @staticmethod - def update_vendor_page( - db: Session, - page_id: int, - vendor_id: int, - title: str | None = None, - content: str | None = None, - content_format: str | None = None, - meta_description: str | None = None, - meta_keywords: str | None = None, - is_published: bool | None = None, - show_in_footer: bool | None = None, - show_in_header: bool | None = None, - show_in_legal: bool | None = None, - display_order: int | None = None, - updated_by: int | None = None, - ) -> ContentPage: - """ - Update a vendor-specific content page with ownership check. - - Raises: - ContentPageNotFoundException: If page not found - UnauthorizedContentPageAccessException: If page doesn't belong to vendor - """ - page = ContentPageService.get_page_by_id_or_raise(db, page_id) - - if page.vendor_id != vendor_id: - raise UnauthorizedContentPageAccessException(action="edit") - - return ContentPageService.update_page( - db, - page_id=page_id, - title=title, - content=content, - content_format=content_format, - meta_description=meta_description, - meta_keywords=meta_keywords, - is_published=is_published, - show_in_footer=show_in_footer, - show_in_header=show_in_header, - show_in_legal=show_in_legal, - display_order=display_order, - updated_by=updated_by, - ) - - @staticmethod - def delete_vendor_page(db: Session, page_id: int, vendor_id: int) -> None: - """ - Delete a vendor-specific content page with ownership check. - - Raises: - ContentPageNotFoundException: If page not found - UnauthorizedContentPageAccessException: If page doesn't belong to vendor - """ - page = ContentPageService.get_page_by_id_or_raise(db, page_id) - - if page.vendor_id != vendor_id: - raise UnauthorizedContentPageAccessException(action="delete") - - ContentPageService.delete_page(db, page_id) - - @staticmethod - def create_vendor_override( - db: Session, - platform_id: int, - vendor_id: int, - slug: str, - title: str, - content: str, - content_format: str = "html", - meta_description: str | None = None, - meta_keywords: str | None = None, - is_published: bool = False, - show_in_footer: bool = True, - show_in_header: bool = False, - show_in_legal: bool = False, - display_order: int = 0, - created_by: int | None = None, - ) -> ContentPage: - """ - Create a vendor override page (vendor-specific customization of a default). - - Args: - db: Database session - platform_id: Platform ID - vendor_id: Vendor ID - slug: Page slug (typically matches a default page) - ... other fields - - Returns: - Created ContentPage - """ - return ContentPageService.create_page( - db, - platform_id=platform_id, - slug=slug, - title=title, - content=content, - vendor_id=vendor_id, - is_platform_page=False, - content_format=content_format, - meta_description=meta_description, - meta_keywords=meta_keywords, - is_published=is_published, - show_in_footer=show_in_footer, - show_in_header=show_in_header, - show_in_legal=show_in_legal, - display_order=display_order, - created_by=created_by, - ) - - @staticmethod - def revert_to_default(db: Session, page_id: int, vendor_id: int) -> None: - """ - Revert a vendor override to the default by deleting the override. - - After deletion, the vendor storefront will use the vendor default page. - - Raises: - ContentPageNotFoundException: If page not found - UnauthorizedContentPageAccessException: If page doesn't belong to vendor - """ - ContentPageService.delete_vendor_page(db, page_id, vendor_id) - - # ========================================================================= - # Admin Methods (for listing all pages) - # ========================================================================= - - @staticmethod - def list_all_pages( - db: Session, - platform_id: int | None = None, - vendor_id: int | None = None, - include_unpublished: bool = False, - page_tier: str | None = None, - ) -> list[ContentPage]: - """ - List all content pages for admin management. - - Args: - db: Database session - platform_id: Optional filter by platform ID - vendor_id: Optional filter by vendor ID - include_unpublished: Include draft pages - page_tier: Optional filter by tier ("platform", "vendor_default", "vendor_override") - - Returns: - List of ContentPage objects - """ - filters = [] - - if platform_id: - filters.append(ContentPage.platform_id == platform_id) - - if vendor_id is not None: - filters.append(ContentPage.vendor_id == vendor_id) - - if not include_unpublished: - filters.append(ContentPage.is_published == True) - - if page_tier == "platform": - filters.append(ContentPage.is_platform_page == True) - filters.append(ContentPage.vendor_id == None) - elif page_tier == "vendor_default": - filters.append(ContentPage.is_platform_page == False) - filters.append(ContentPage.vendor_id == None) - elif page_tier == "vendor_override": - filters.append(ContentPage.vendor_id != None) - - return ( - db.query(ContentPage) - .filter(and_(*filters) if filters else True) - .order_by( - ContentPage.platform_id, - ContentPage.vendor_id, - ContentPage.display_order, - ContentPage.title, - ) - .all() - ) - - @staticmethod - def list_all_vendor_pages( - db: Session, - vendor_id: int, - platform_id: int | None = None, - include_unpublished: bool = False, - ) -> list[ContentPage]: - """ - List only vendor-specific pages (overrides and custom pages). - - Args: - db: Database session - vendor_id: Vendor ID - platform_id: Optional filter by platform - include_unpublished: Include draft pages - - Returns: - List of vendor-specific ContentPage objects - """ - filters = [ContentPage.vendor_id == vendor_id] - - if platform_id: - filters.append(ContentPage.platform_id == platform_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() - ) - - # ========================================================================= - # Helper Methods for raising exceptions - # ========================================================================= - - @staticmethod - def update_page_or_raise( - db: Session, - page_id: int, - title: str | None = None, - content: str | None = None, - content_format: str | None = None, - template: str | None = None, - meta_description: str | None = None, - meta_keywords: str | None = None, - is_published: bool | None = None, - show_in_footer: bool | None = None, - show_in_header: bool | None = None, - show_in_legal: bool | None = None, - display_order: int | None = None, - updated_by: int | None = None, - ) -> ContentPage: - """ - Update an existing content page or raise exception. - - Raises: - ContentPageNotFoundException: If page not found - """ - page = ContentPageService.update_page( - db, - page_id=page_id, - title=title, - content=content, - content_format=content_format, - template=template, - meta_description=meta_description, - meta_keywords=meta_keywords, - is_published=is_published, - show_in_footer=show_in_footer, - show_in_header=show_in_header, - show_in_legal=show_in_legal, - display_order=display_order, - updated_by=updated_by, - ) - if not page: - raise ContentPageNotFoundException(identifier=page_id) - return page - - @staticmethod - def delete_page_or_raise(db: Session, page_id: int) -> None: - """ - Delete a content page or raise exception. - - Raises: - ContentPageNotFoundException: If page not found - """ - success = ContentPageService.delete_page(db, page_id) - if not success: - raise ContentPageNotFoundException(identifier=page_id) - - # ========================================================================= - # Homepage Sections Management - # ========================================================================= - - @staticmethod - def update_homepage_sections( - db: Session, - page_id: int, - sections: dict, - updated_by: int | None = None, - ) -> ContentPage: - """ - Update homepage sections with validation. - - Args: - db: Database session - page_id: Content page ID - sections: Homepage sections dict (validated against HomepageSections schema) - updated_by: User ID making the update - - Returns: - Updated ContentPage - - Raises: - ContentPageNotFoundException: If page not found - ValidationError: If sections schema invalid - """ - from models.schema.homepage_sections import HomepageSections - - page = ContentPageService.get_page_by_id_or_raise(db, page_id) - - # Validate sections against schema - validated = HomepageSections(**sections) - - # Update page - page.sections = validated.model_dump() - page.updated_by = updated_by - - db.flush() - db.refresh(page) - - logger.info(f"[CMS] Updated homepage sections for page_id={page_id}") - return page - - @staticmethod - def update_single_section( - db: Session, - page_id: int, - section_name: str, - section_data: dict, - updated_by: int | None = None, - ) -> ContentPage: - """ - Update a single section within homepage sections. - - Args: - db: Database session - page_id: Content page ID - section_name: Section to update (hero, features, pricing, cta) - section_data: Section configuration dict - updated_by: User ID making the update - - Returns: - Updated ContentPage - - Raises: - ContentPageNotFoundException: If page not found - ValueError: If section name is invalid - """ - from models.schema.homepage_sections import ( - HeroSection, - FeaturesSection, - PricingSection, - CTASection, - ) - - SECTION_SCHEMAS = { - "hero": HeroSection, - "features": FeaturesSection, - "pricing": PricingSection, - "cta": CTASection, - } - - if section_name not in SECTION_SCHEMAS: - raise ValueError(f"Invalid section name: {section_name}. Must be one of: {list(SECTION_SCHEMAS.keys())}") - - page = ContentPageService.get_page_by_id_or_raise(db, page_id) - - # Validate section data against its schema - schema = SECTION_SCHEMAS[section_name] - validated_section = schema(**section_data) - - # Initialize sections if needed - current_sections = page.sections or {} - current_sections[section_name] = validated_section.model_dump() - - page.sections = current_sections - page.updated_by = updated_by - - db.flush() - db.refresh(page) - - logger.info(f"[CMS] Updated section '{section_name}' for page_id={page_id}") - return page - - @staticmethod - def get_default_sections(languages: list[str] | None = None) -> dict: - """ - Get empty sections structure for new homepage. - - Args: - languages: List of language codes from platform.supported_languages. - Defaults to ['fr', 'de', 'en'] if not provided. - - Returns: - Empty sections dict with language placeholders - """ - from models.schema.homepage_sections import HomepageSections - - if languages is None: - languages = ["fr", "de", "en"] - - return HomepageSections.get_empty_structure(languages).model_dump() - - -# Singleton instance -content_page_service = ContentPageService() +__all__ = ["ContentPageService", "content_page_service"] diff --git a/app/templates_config.py b/app/templates_config.py index cd528ab7..b327a04b 100644 --- a/app/templates_config.py +++ b/app/templates_config.py @@ -4,10 +4,21 @@ Shared Jinja2 templates configuration. All route modules should import `templates` from here to ensure consistent globals (like translation function) are available. + +Template Loading Strategy: +- Core templates from app/templates/ (highest priority) +- Module templates from app/modules//templates/ (namespaced) + +Module templates should use namespace prefix to avoid collisions: + app/modules/cms/templates/cms/admin/pages.html + -> Rendered as: templates.TemplateResponse("cms/admin/pages.html", ...) """ + +import logging from pathlib import Path from fastapi.templating import Jinja2Templates +from jinja2 import ChoiceLoader, FileSystemLoader from app.utils.i18n import ( LANGUAGE_FLAGS, @@ -17,11 +28,60 @@ from app.utils.i18n import ( create_translation_context, ) -# Templates directory -TEMPLATES_DIR = Path(__file__).parent / "templates" +logger = logging.getLogger(__name__) -# Create shared templates instance +# Core templates directory +TEMPLATES_DIR = Path(__file__).parent / "templates" +MODULES_DIR = Path(__file__).parent / "modules" + + +def create_template_loaders() -> ChoiceLoader: + """ + Create a ChoiceLoader that searches multiple template directories. + + Search order: + 1. Core templates (app/templates/) - highest priority + 2. Module templates (app/modules//templates/) - in alphabetical order + + Returns: + ChoiceLoader configured with all template directories + """ + loaders = [FileSystemLoader(str(TEMPLATES_DIR))] # Core templates first + + # Add module template directories + if MODULES_DIR.exists(): + for module_dir in sorted(MODULES_DIR.iterdir()): + if module_dir.is_dir(): + templates_path = module_dir / "templates" + if templates_path.exists() and templates_path.is_dir(): + loaders.append(FileSystemLoader(str(templates_path))) + logger.debug(f"[Templates] Added module templates: {module_dir.name}") + + return ChoiceLoader(loaders) + + +def get_module_template_dirs() -> list[Path]: + """ + Get list of all module template directories. + + Useful for debugging and introspection. + + Returns: + List of Path objects for module template directories + """ + dirs = [] + if MODULES_DIR.exists(): + for module_dir in sorted(MODULES_DIR.iterdir()): + if module_dir.is_dir(): + templates_path = module_dir / "templates" + if templates_path.exists() and templates_path.is_dir(): + dirs.append(templates_path) + return dirs + + +# Create shared templates instance with multi-directory loader templates = Jinja2Templates(directory=str(TEMPLATES_DIR)) +templates.env.loader = create_template_loaders() # Add translation function to Jinja2 environment globals # This makes _() available in all templates AND macros diff --git a/models/database/content_page.py b/models/database/content_page.py index 93aa9bca..03e8ac30 100644 --- a/models/database/content_page.py +++ b/models/database/content_page.py @@ -24,6 +24,9 @@ Features: - SEO metadata - Published/Draft status - Navigation placement (header, footer, legal) + +NOTE: This is the canonical location for the ContentPage model. +The CMS module (app.modules.cms) re-exports this model. """ from datetime import UTC, datetime @@ -230,8 +233,3 @@ class ContentPage(Base): "created_by": self.created_by, "updated_by": self.updated_by, } - - -# Add relationship to Vendor model -# This should be added to models/database/vendor.py: -# content_pages = relationship("ContentPage", back_populates="vendor", cascade="all, delete-orphan")