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

feat: add multi-platform CMS architecture (Phase 1)

Browse Source 
Implement the foundation for multi-platform support allowing independent
business offerings (OMS, Loyalty, etc.) with their own CMS pages.

Database Models:
- Add Platform model for business offerings (domain, branding, config)
- Add VendorPlatform junction table for many-to-many relationship
- Update SubscriptionTier with platform_id and CMS limits
- Update ContentPage with platform_id, is_platform_page for three-tier hierarchy
- Add CMS feature codes (cms_basic, cms_custom_pages, cms_templates, etc.)

Three-Tier Content Resolution:
1. Vendor override (platform_id + vendor_id + slug)
2. Vendor default (platform_id + vendor_id=NULL + is_platform_page=False)
3. Platform marketing pages (is_platform_page=True)

New Components:
- PlatformContextMiddleware for detecting platform from domain/path
- ContentPageService updated with full three-tier resolution
- Platform folder structure (app/platforms/oms/, app/platforms/loyalty/)
- Alembic migration with backfill for existing data

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-01-18 19:49:44 +01:00
parent 4c9b3c4e4b
commit 408019dbb3
24 changed files with 2049 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

374
alembic/versions/z4e5f6a7b8c9_add_multi_platform_support.py Normal file
View File

@@ -0,0 +1,374 @@
"""add multi-platform support
Revision ID: z4e5f6a7b8c9
Revises: 1b398cf45e85
Create Date: 2026-01-18 12:00:00.000000
This migration adds multi-platform support:
1. Creates platforms table for business offerings (OMS, Loyalty, etc.)
2. Creates vendor_platforms junction table for many-to-many relationship
3. Adds platform_id and CMS limits to subscription_tiers
4. Adds platform_id and is_platform_page to content_pages
5. Inserts default "oms" platform
6. Backfills existing data to OMS platform
7. Creates vendor_platforms entries for existing vendors
"""
import json
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "z4e5f6a7b8c9"
down_revision: Union[str, None] = "1b398cf45e85"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
# Platform marketing page slugs (is_platform_page=True)
PLATFORM_PAGE_SLUGS = [
"platform_homepage",
"home",
"pricing",
"about",
"contact",
"faq",
"terms",
"privacy",
"features",
"integrations",
]
# CMS limits per tier
CMS_TIER_LIMITS = {
"essential": {"cms_pages_limit": 3, "cms_custom_pages_limit": 0},
"professional": {"cms_pages_limit": 10, "cms_custom_pages_limit": 5},
"business": {"cms_pages_limit": 30, "cms_custom_pages_limit": 20},
"enterprise": {"cms_pages_limit": None, "cms_custom_pages_limit": None}, # Unlimited
}
# CMS features per tier
CMS_TIER_FEATURES = {
"essential": ["cms_basic"],
"professional": ["cms_basic", "cms_custom_pages", "cms_seo"],
"business": ["cms_basic", "cms_custom_pages", "cms_seo", "cms_templates"],
"enterprise": ["cms_basic", "cms_custom_pages", "cms_unlimited_pages", "cms_templates", "cms_seo", "cms_scheduling"],
}
def upgrade() -> None:
conn = op.get_bind()
# =========================================================================
# 1. Create platforms table
# =========================================================================
op.create_table(
"platforms",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("code", sa.String(50), nullable=False),
sa.Column("name", sa.String(100), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("domain", sa.String(255), nullable=True),
sa.Column("path_prefix", sa.String(50), nullable=True),
sa.Column("logo", sa.String(500), nullable=True),
sa.Column("logo_dark", sa.String(500), nullable=True),
sa.Column("favicon", sa.String(500), nullable=True),
sa.Column("theme_config", sa.JSON(), nullable=True),
sa.Column("default_language", sa.String(5), nullable=False, server_default="fr"),
sa.Column("supported_languages", sa.JSON(), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default="true"),
sa.Column("is_public", sa.Boolean(), nullable=False, server_default="true"),
sa.Column("settings", sa.JSON(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_platforms_code", "platforms", ["code"], unique=True)
op.create_index("ix_platforms_domain", "platforms", ["domain"], unique=True)
op.create_index("ix_platforms_path_prefix", "platforms", ["path_prefix"], unique=True)
op.create_index("idx_platform_active", "platforms", ["is_active"])
op.create_index("idx_platform_public", "platforms", ["is_public", "is_active"])
# =========================================================================
# 2. Create vendor_platforms junction table
# =========================================================================
op.create_table(
"vendor_platforms",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("vendor_id", sa.Integer(), nullable=False),
sa.Column("platform_id", sa.Integer(), nullable=False),
sa.Column("tier_id", sa.Integer(), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default="true"),
sa.Column("is_primary", sa.Boolean(), nullable=False, server_default="false"),
sa.Column("custom_subdomain", sa.String(100), nullable=True),
sa.Column("settings", sa.JSON(), nullable=True),
sa.Column("joined_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
sa.ForeignKeyConstraint(["vendor_id"], ["vendors.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["platform_id"], ["platforms.id"], ondelete="CASCADE"),
sa.ForeignKeyConstraint(["tier_id"], ["subscription_tiers.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_vendor_platforms_vendor_id", "vendor_platforms", ["vendor_id"])
op.create_index("ix_vendor_platforms_platform_id", "vendor_platforms", ["platform_id"])
op.create_index("ix_vendor_platforms_tier_id", "vendor_platforms", ["tier_id"])
op.create_index("idx_vendor_platform_active", "vendor_platforms", ["vendor_id", "platform_id", "is_active"])
op.create_index("idx_vendor_platform_primary", "vendor_platforms", ["vendor_id", "is_primary"])
op.create_unique_constraint("uq_vendor_platform", "vendor_platforms", ["vendor_id", "platform_id"])
# =========================================================================
# 3. Add platform_id and CMS columns to subscription_tiers
# =========================================================================
# Add platform_id column (nullable for global tiers)
op.add_column(
"subscription_tiers",
sa.Column("platform_id", sa.Integer(), nullable=True),
)
op.create_index("ix_subscription_tiers_platform_id", "subscription_tiers", ["platform_id"])
op.create_foreign_key(
"fk_subscription_tiers_platform_id",
"subscription_tiers",
"platforms",
["platform_id"],
["id"],
ondelete="CASCADE",
)
# Add CMS limit columns
op.add_column(
"subscription_tiers",
sa.Column("cms_pages_limit", sa.Integer(), nullable=True),
)
op.add_column(
"subscription_tiers",
sa.Column("cms_custom_pages_limit", sa.Integer(), nullable=True),
)
op.create_index("idx_tier_platform_active", "subscription_tiers", ["platform_id", "is_active"])
# =========================================================================
# 4. Add platform_id and is_platform_page to content_pages
# =========================================================================
# Add platform_id column (will be set to NOT NULL after backfill)
op.add_column(
"content_pages",
sa.Column("platform_id", sa.Integer(), nullable=True),
)
op.create_index("ix_content_pages_platform_id", "content_pages", ["platform_id"])
# Add is_platform_page column
op.add_column(
"content_pages",
sa.Column("is_platform_page", sa.Boolean(), nullable=False, server_default="false"),
)
op.create_index("idx_platform_page_type", "content_pages", ["platform_id", "is_platform_page"])
# =========================================================================
# 5. Insert default OMS platform
# =========================================================================
conn.execute(
sa.text("""
INSERT INTO platforms (code, name, description, domain, path_prefix, default_language,
supported_languages, is_active, is_public, theme_config, settings,
created_at, updated_at)
VALUES ('oms', 'Wizamart OMS', 'Order Management System for Luxembourg merchants',
'oms.lu', 'oms', 'fr', '["fr", "de", "en"]', true, true, '{}', '{}',
CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
""")
)
# Get the OMS platform ID
result = conn.execute(sa.text("SELECT id FROM platforms WHERE code = 'oms'"))
oms_platform_id = result.fetchone()[0]
# =========================================================================
# 6. Backfill content_pages with platform_id
# =========================================================================
conn.execute(
sa.text(f"UPDATE content_pages SET platform_id = {oms_platform_id} WHERE platform_id IS NULL")
)
# Set is_platform_page=True for platform marketing page slugs
# Only for pages that have vendor_id=NULL (platform-level pages)
slugs_list = ", ".join([f"'{slug}'" for slug in PLATFORM_PAGE_SLUGS])
conn.execute(
sa.text(f"""
UPDATE content_pages
SET is_platform_page = true
WHERE vendor_id IS NULL AND slug IN ({slugs_list})
""")
)
# Make platform_id NOT NULL after backfill
op.alter_column("content_pages", "platform_id", nullable=False)
# Add foreign key constraint
op.create_foreign_key(
"fk_content_pages_platform_id",
"content_pages",
"platforms",
["platform_id"],
["id"],
ondelete="CASCADE",
)
# =========================================================================
# 7. Update content_pages constraints
# =========================================================================
# Drop old unique constraint
op.drop_constraint("uq_vendor_slug", "content_pages", type_="unique")
# Create new unique constraint including platform_id
op.create_unique_constraint(
"uq_platform_vendor_slug",
"content_pages",
["platform_id", "vendor_id", "slug"],
)
# Update indexes
op.drop_index("idx_vendor_published", table_name="content_pages")
op.drop_index("idx_slug_published", table_name="content_pages")
op.create_index("idx_platform_vendor_published", "content_pages", ["platform_id", "vendor_id", "is_published"])
op.create_index("idx_platform_slug_published", "content_pages", ["platform_id", "slug", "is_published"])
# =========================================================================
# 8. Update subscription_tiers with CMS limits
# =========================================================================
for tier_code, limits in CMS_TIER_LIMITS.items():
cms_pages = limits["cms_pages_limit"] if limits["cms_pages_limit"] is not None else "NULL"
cms_custom = limits["cms_custom_pages_limit"] if limits["cms_custom_pages_limit"] is not None else "NULL"
conn.execute(
sa.text(f"""
UPDATE subscription_tiers
SET cms_pages_limit = {cms_pages},
cms_custom_pages_limit = {cms_custom}
WHERE code = '{tier_code}'
""")
)
# Add CMS features to each tier
for tier_code, cms_features in CMS_TIER_FEATURES.items():
# Get current features
result = conn.execute(
sa.text(f"SELECT features FROM subscription_tiers WHERE code = '{tier_code}'")
)
row = result.fetchone()
if row and row[0]:
current_features = json.loads(row[0]) if isinstance(row[0], str) else row[0]
else:
current_features = []
# Add CMS features that aren't already present
for feature in cms_features:
if feature not in current_features:
current_features.append(feature)
# Update features
features_json = json.dumps(current_features)
conn.execute(
sa.text(f"UPDATE subscription_tiers SET features = '{features_json}' WHERE code = '{tier_code}'")
)
# =========================================================================
# 9. Create vendor_platforms entries for existing vendors
# =========================================================================
# Get all vendors with their subscription tier_id
conn.execute(
sa.text(f"""
INSERT INTO vendor_platforms (vendor_id, platform_id, tier_id, is_active, is_primary,
joined_at, created_at, updated_at)
SELECT v.id, {oms_platform_id}, vs.tier_id, v.is_active, true,
v.created_at, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
FROM vendors v
LEFT JOIN vendor_subscriptions vs ON vs.vendor_id = v.id
""")
)
# =========================================================================
# 10. Add CMS feature records to features table
# =========================================================================
# Get minimum tier IDs for CMS features
tier_ids = {}
result = conn.execute(sa.text("SELECT id, code FROM subscription_tiers"))
for row in result:
tier_ids[row[1]] = row[0]
cms_features = [
("cms", "cms_basic", "Basic CMS", "Override default pages with custom content", "settings", "document-text", None, tier_ids.get("essential"), 1),
("cms", "cms_custom_pages", "Custom Pages", "Create custom pages beyond defaults", "settings", "document-add", None, tier_ids.get("professional"), 2),
("cms", "cms_unlimited_pages", "Unlimited Pages", "No page limit", "settings", "documents", None, tier_ids.get("enterprise"), 3),
("cms", "cms_templates", "Page Templates", "Access to page templates", "settings", "template", None, tier_ids.get("business"), 4),
("cms", "cms_seo", "Advanced SEO", "SEO metadata and optimization", "settings", "search", None, tier_ids.get("professional"), 5),
("cms", "cms_scheduling", "Page Scheduling", "Schedule page publish/unpublish", "settings", "clock", None, tier_ids.get("enterprise"), 6),
]
for category, code, name, description, ui_location, ui_icon, ui_route, minimum_tier_id, display_order in cms_features:
min_tier_val = minimum_tier_id if minimum_tier_id else "NULL"
conn.execute(
sa.text(f"""
INSERT INTO features (code, name, description, category, ui_location, ui_icon, ui_route,
minimum_tier_id, is_active, is_visible, display_order, created_at, updated_at)
VALUES ('{code}', '{name}', '{description}', '{category}', '{ui_location}', '{ui_icon}', NULL,
{min_tier_val}, true, true, {display_order}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT (code) DO NOTHING
""")
)
def downgrade() -> None:
conn = op.get_bind()
# Remove CMS features from features table
conn.execute(
sa.text("""
DELETE FROM features
WHERE code IN ('cms_basic', 'cms_custom_pages', 'cms_unlimited_pages',
'cms_templates', 'cms_seo', 'cms_scheduling')
""")
)
# Remove CMS features from subscription_tiers
result = conn.execute(sa.text("SELECT id, code, features FROM subscription_tiers"))
for row in result:
tier_id, tier_code, features_data = row
if features_data:
current_features = json.loads(features_data) if isinstance(features_data, str) else features_data
# Remove CMS features
updated_features = [f for f in current_features if not f.startswith("cms_")]
features_json = json.dumps(updated_features)
conn.execute(
sa.text(f"UPDATE subscription_tiers SET features = '{features_json}' WHERE id = {tier_id}")
)
# Drop vendor_platforms table
op.drop_table("vendor_platforms")
# Restore old content_pages constraints
op.drop_constraint("uq_platform_vendor_slug", "content_pages", type_="unique")
op.drop_index("idx_platform_vendor_published", table_name="content_pages")
op.drop_index("idx_platform_slug_published", table_name="content_pages")
op.drop_index("idx_platform_page_type", table_name="content_pages")
op.drop_constraint("fk_content_pages_platform_id", "content_pages", type_="foreignkey")
op.drop_index("ix_content_pages_platform_id", table_name="content_pages")
op.drop_column("content_pages", "is_platform_page")
op.drop_column("content_pages", "platform_id")
op.create_unique_constraint("uq_vendor_slug", "content_pages", ["vendor_id", "slug"])
op.create_index("idx_vendor_published", "content_pages", ["vendor_id", "is_published"])
op.create_index("idx_slug_published", "content_pages", ["slug", "is_published"])
# Remove subscription_tiers platform columns
op.drop_index("idx_tier_platform_active", table_name="subscription_tiers")
op.drop_constraint("fk_subscription_tiers_platform_id", "subscription_tiers", type_="foreignkey")
op.drop_index("ix_subscription_tiers_platform_id", table_name="subscription_tiers")
op.drop_column("subscription_tiers", "cms_custom_pages_limit")
op.drop_column("subscription_tiers", "cms_pages_limit")
op.drop_column("subscription_tiers", "platform_id")
# Drop platforms table
op.drop_index("idx_platform_public", table_name="platforms")
op.drop_index("idx_platform_active", table_name="platforms")
op.drop_index("ix_platforms_path_prefix", table_name="platforms")
op.drop_index("ix_platforms_domain", table_name="platforms")
op.drop_index("ix_platforms_code", table_name="platforms")
op.drop_table("platforms")

15
app/platforms/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
# app/platforms/__init__.py
"""
Platform-specific code and configurations.
Each platform (OMS, Loyalty, etc.) has its own:
- routes/: Platform-specific routes
- templates/: Platform-specific templates
- config.py: Platform configuration
Shared code that applies to all platforms lives in shared/.
"""
from .shared.base_platform import BasePlatformConfig
__all__ = ["BasePlatformConfig"]

10
app/platforms/loyalty/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
# app/platforms/loyalty/__init__.py
"""
Loyalty Platform
Platform for customer loyalty programs and rewards.
"""
from .config import LoyaltyPlatformConfig
__all__ = ["LoyaltyPlatformConfig"]

49
app/platforms/loyalty/config.py Normal file
View File

@@ -0,0 +1,49 @@
# app/platforms/loyalty/config.py
"""
Loyalty Platform Configuration
Configuration for the Loyalty/Rewards platform.
"""
from app.platforms.shared.base_platform import BasePlatformConfig
class LoyaltyPlatformConfig(BasePlatformConfig):
"""Configuration for the Loyalty platform."""
@property
def code(self) -> str:
return "loyalty"
@property
def name(self) -> str:
return "Loyalty+"
@property
def description(self) -> str:
return "Customer loyalty and rewards platform"
@property
def features(self) -> list[str]:
"""Loyalty-specific features."""
return [
"loyalty_points",
"rewards_catalog",
"customer_tiers",
"referral_program",
]
@property
def vendor_default_page_slugs(self) -> list[str]:
"""Default pages for Loyalty vendor storefronts."""
return [
"about",
"how-it-works",
"rewards",
"terms-of-service",
"privacy-policy",
]
# Singleton instance
loyalty_config = LoyaltyPlatformConfig()

2
app/platforms/loyalty/routes/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/loyalty/routes/__init__.py
"""Loyalty platform routes."""

2
app/platforms/loyalty/templates/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/loyalty/templates/__init__.py
"""Loyalty platform templates."""

10
app/platforms/oms/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
# app/platforms/oms/__init__.py
"""
OMS (Order Management System) Platform
The primary platform for managing orders, products, and vendor storefronts.
"""
from .config import OMSPlatformConfig
__all__ = ["OMSPlatformConfig"]

52
app/platforms/oms/config.py Normal file
View File

@@ -0,0 +1,52 @@
# app/platforms/oms/config.py
"""
OMS Platform Configuration
Configuration for the Order Management System platform.
"""
from app.platforms.shared.base_platform import BasePlatformConfig
class OMSPlatformConfig(BasePlatformConfig):
"""Configuration for the OMS platform."""
@property
def code(self) -> str:
return "oms"
@property
def name(self) -> str:
return "Wizamart OMS"
@property
def description(self) -> str:
return "Order Management System for Luxembourg merchants"
@property
def features(self) -> list[str]:
"""OMS-specific features."""
return [
"order_management",
"inventory_basic",
"invoice_lu",
"letzshop_sync",
"customer_view",
]
@property
def vendor_default_page_slugs(self) -> list[str]:
"""Default pages for OMS vendor storefronts."""
return [
"about",
"shipping",
"returns",
"privacy-policy",
"terms-of-service",
"contact",
"faq",
]
# Singleton instance
oms_config = OMSPlatformConfig()

2
app/platforms/oms/routes/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/oms/routes/__init__.py
"""OMS platform routes."""

2
app/platforms/oms/templates/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/oms/templates/__init__.py
"""OMS platform templates."""

6
app/platforms/shared/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
# app/platforms/shared/__init__.py
"""Shared platform code and base classes."""
from .base_platform import BasePlatformConfig
__all__ = ["BasePlatformConfig"]

97
app/platforms/shared/base_platform.py Normal file
View File

@@ -0,0 +1,97 @@
# app/platforms/shared/base_platform.py
"""
Base Platform Configuration
Provides a base class for platform-specific configurations.
Each platform (OMS, Loyalty, etc.) should extend this class.
"""
from abc import ABC, abstractmethod
from typing import Any
class BasePlatformConfig(ABC):
"""
Base configuration class for platforms.
Each platform should create a config.py that extends this class
and provides platform-specific settings.
"""
@property
@abstractmethod
def code(self) -> str:
"""Unique platform code (e.g., 'oms', 'loyalty')."""
pass
@property
@abstractmethod
def name(self) -> str:
"""Display name (e.g., 'Wizamart OMS')."""
pass
@property
def description(self) -> str:
"""Platform description."""
return ""
@property
def default_language(self) -> str:
"""Default language code."""
return "fr"
@property
def supported_languages(self) -> list[str]:
"""List of supported language codes."""
return ["fr", "de", "en"]
@property
def theme_defaults(self) -> dict[str, Any]:
"""Default theme configuration."""
return {
"primary_color": "#6366f1",
"secondary_color": "#8b5cf6",
"font_family": "Inter, sans-serif",
}
@property
def features(self) -> list[str]:
"""List of feature codes enabled for this platform."""
return []
@property
def marketing_page_slugs(self) -> list[str]:
"""
Slugs that should be treated as platform marketing pages.
These pages describe the platform itself (pricing, features, etc.)
rather than being vendor storefront content.
"""
return [
"home",
"platform_homepage",
"pricing",
"about",
"contact",
"faq",
"terms",
"privacy",
"features",
"integrations",
]
@property
def vendor_default_page_slugs(self) -> list[str]:
"""
Slugs for default vendor storefront pages.
These pages provide fallback content for vendors who haven't
customized their storefront.
"""
return [
"about",
"shipping",
"returns",
"privacy-policy",
"terms-of-service",
]

2
app/platforms/shared/routes/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/shared/routes/__init__.py
"""Shared platform routes."""

2
app/platforms/shared/templates/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
# app/platforms/shared/templates/__init__.py
"""Shared platform templates."""

733
app/services/content_page_service.py
View File

@@ -2,18 +2,24 @@
"""
Content Page Service
Business logic for managing content pages with platform defaults
and vendor-specific overrides.
Business logic for managing content pages with three-tier hierarchy:
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
1. Platform Marketing Pages (is_platform_page=True, vendor_id=NULL)
- Platform's own pages (homepage, pricing, about)
- Describe the platform/business offering itself
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
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
@@ -32,67 +38,124 @@ logger = logging.getLogger(__name__)
class ContentPageService:
"""Service for content page operations."""
"""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 fallback to platform default.
Get content page for a vendor with three-tier resolution.
Lookup order:
1. Vendor-specific override (vendor_id + slug)
2. Platform default (vendor_id=NULL + slug)
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 platform defaults only)
vendor_id: Vendor ID (None for defaults only)
include_unpublished: Include draft pages (for preview)
Returns:
ContentPage or None
"""
filters = [ContentPage.slug == slug]
base_filters = [
ContentPage.platform_id == platform_id,
ContentPage.slug == slug,
]
if not include_unpublished:
filters.append(ContentPage.is_published == True)
base_filters.append(ContentPage.is_published == True)
# Try vendor-specific override first
# Tier 1: Try vendor-specific override first
if vendor_id:
vendor_page = (
db.query(ContentPage)
.filter(and_(ContentPage.vendor_id == vendor_id, *filters))
.filter(and_(ContentPage.vendor_id == vendor_id, *base_filters))
.first()
)
if vendor_page:
logger.debug(
f"Found vendor-specific page: {slug} for vendor_id={vendor_id}"
f"[CMS] Found vendor override: {slug} for vendor_id={vendor_id}, platform_id={platform_id}"
)
return vendor_page
# Fallback to platform default
platform_page = (
# Tier 2: Fallback to vendor default (not platform page)
vendor_default_page = (
db.query(ContentPage)
.filter(and_(ContentPage.vendor_id == None, *filters))
.filter(
and_(
ContentPage.vendor_id == None,
ContentPage.is_platform_page == False,
*base_filters,
)
)
.first()
)
if platform_page:
logger.debug(f"Using platform default page: {slug}")
else:
logger.debug(f"No page found for slug: {slug}")
if vendor_default_page:
logger.debug(f"[CMS] Using vendor default page: {slug} for platform_id={platform_id}")
return vendor_default_page
return platform_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,
@@ -100,13 +163,15 @@ class ContentPageService:
legal_only: bool = False,
) -> list[ContentPage]:
"""
List all available pages for a vendor (includes vendor overrides + platform defaults).
List all available pages for a vendor storefront.
Merges vendor-specific overrides with platform defaults, prioritizing vendor overrides.
Merges vendor overrides with vendor defaults, prioritizing overrides.
Does NOT include platform marketing pages.
Args:
db: Database session
vendor_id: Vendor ID (None for platform pages only)
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
@@ -115,7 +180,81 @@ class ContentPageService:
Returns:
List of ContentPage objects
"""
filters = []
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)
@@ -126,45 +265,59 @@ class ContentPageService:
if header_only:
filters.append(ContentPage.show_in_header == True)
if legal_only:
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, *filters))
.order_by(ContentPage.display_order, ContentPage.title)
.all()
)
# Get platform defaults
platform_pages = (
return (
db.query(ContentPage)
.filter(and_(ContentPage.vendor_id == None, *filters))
.filter(and_(*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
@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,
]
# Sort by display_order
all_pages.sort(key=lambda p: (p.display_order, p.title))
if not include_unpublished:
filters.append(ContentPage.is_published == True)
return all_pages
return (
db.query(ContentPage)
.filter(and_(*filters))
.order_by(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,
@@ -181,12 +334,14 @@ class ContentPageService:
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)
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 homepage/landing pages (default, minimal, modern, etc.)
template: Template name for landing pages
meta_description: SEO description
meta_keywords: SEO keywords
is_published: Publish immediately
@@ -200,7 +355,9 @@ class ContentPageService:
Created ContentPage
"""
page = ContentPage(
platform_id=platform_id,
vendor_id=vendor_id,
is_platform_page=is_platform_page,
slug=slug,
title=title,
content=content,
@@ -222,8 +379,9 @@ class ContentPageService:
db.flush()
db.refresh(page)
page_type = "platform" if is_platform_page else ("vendor" if vendor_id else "default")
logger.info(
f"Created content page: {slug} (vendor_id={vendor_id}, id={page.id})"
f"[CMS] Created {page_type} page: {slug} (platform_id={platform_id}, vendor_id={vendor_id}, id={page.id})"
)
return page
@@ -250,18 +408,7 @@ class ContentPageService:
Args:
db: Database session
page_id: Page ID
title: New title
content: New content
content_format: New format
template: New template name
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
show_in_legal: New legal bar visibility
display_order: New sort order
updated_by: User ID who updated it
... other fields
Returns:
Updated ContentPage or None if not found
@@ -269,7 +416,7 @@ class ContentPageService:
page = db.query(ContentPage).filter(ContentPage.id == page_id).first()
if not page:
logger.warning(f"Content page not found: id={page_id}")
logger.warning(f"[CMS] Content page not found: id={page_id}")
return None
# Update fields if provided
@@ -303,7 +450,7 @@ class ContentPageService:
db.flush()
db.refresh(page)
logger.info(f"Updated content page: id={page_id}, slug={page.slug}")
logger.info(f"[CMS] Updated content page: id={page_id}, slug={page.slug}")
return page
@staticmethod
@@ -321,12 +468,12 @@ class ContentPageService:
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}")
logger.warning(f"[CMS] Content page not found for deletion: id={page_id}")
return False
db.delete(page)
logger.info(f"Deleted content page: id={page_id}, slug={page.slug}")
logger.info(f"[CMS] Deleted content page: id={page_id}, slug={page.slug}")
return True
@staticmethod
@@ -338,16 +485,6 @@ class ContentPageService:
def get_page_by_id_or_raise(db: Session, page_id: int) -> ContentPage:
"""
Get content page by ID or raise ContentPageNotFoundException.
Args:
db: Database session
page_id: Page ID
Returns:
ContentPage
Raises:
ContentPageNotFoundException: If page not found
"""
page = db.query(ContentPage).filter(ContentPage.id == page_id).first()
if not page:
@@ -357,33 +494,269 @@ class ContentPageService:
@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 fallback to platform default.
Get content page for a vendor with three-tier resolution.
Raises ContentPageNotFoundException if not found.
Args:
db: Database session
slug: Page slug
vendor_id: Vendor ID
include_unpublished: Include draft pages
Returns:
ContentPage
Raises:
ContentPageNotFoundException: If page not found
"""
page = ContentPageService.get_page_for_vendor(
db, slug=slug, vendor_id=vendor_id, include_unpublished=include_unpublished
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,
@@ -439,168 +812,6 @@ class ContentPageService:
if not success:
raise ContentPageNotFoundException(identifier=page_id)
@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.
Args:
db: Database session
page_id: Page ID
vendor_id: Vendor ID (for ownership verification)
... other fields
Returns:
Updated ContentPage
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_or_raise(
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.
Args:
db: Database session
page_id: Page ID
vendor_id: Vendor ID (for ownership verification)
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_or_raise(db, page_id)
@staticmethod
def list_all_pages(
db: Session,
vendor_id: int | None = None,
include_unpublished: bool = False,
) -> list[ContentPage]:
"""
List all content pages (platform defaults and vendor overrides).
Args:
db: Database session
vendor_id: Optional filter by vendor ID
include_unpublished: Include draft pages
Returns:
List of ContentPage objects
"""
filters = []
if vendor_id:
filters.append(ContentPage.vendor_id == vendor_id)
if not include_unpublished:
filters.append(ContentPage.is_published == True)
return (
db.query(ContentPage)
.filter(and_(*filters) if filters else True)
.order_by(
ContentPage.vendor_id, ContentPage.display_order, ContentPage.title
)
.all()
)
@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()

27
middleware/__init__.py
View File

@@ -1 +1,28 @@
# middleware/__init__.py
"""Middleware package for request processing."""
from .platform_context import (
PlatformContextManager,
PlatformContextMiddleware,
get_current_platform,
require_platform_context,
)
from .vendor_context import (
VendorContextManager,
VendorContextMiddleware,
get_current_vendor,
require_vendor_context,
)
__all__ = [
# Platform context
"PlatformContextManager",
"PlatformContextMiddleware",
"get_current_platform",
"require_platform_context",
# Vendor context
"VendorContextManager",
"VendorContextMiddleware",
"get_current_vendor",
"require_vendor_context",
]

375
middleware/platform_context.py Normal file
View File

@@ -0,0 +1,375 @@
# middleware/platform_context.py
"""
Platform Context Middleware
Detects platform from host/domain/path and injects into request.state.
This middleware runs BEFORE VendorContextMiddleware to establish platform context.
Handles two routing modes:
1. Production: Domain-based (oms.lu, loyalty.lu → Platform detection)
2. Development: Path-based (localhost:9999/oms/*, localhost:9999/loyalty/* → Platform detection)
Also provides platform_clean_path for downstream middleware to use.
"""
import logging
from fastapi import Request
from sqlalchemy.orm import Session
from starlette.middleware.base import BaseHTTPMiddleware
from app.core.config import settings
from app.core.database import get_db
from models.database.platform import Platform
logger = logging.getLogger(__name__)
# Default platform code for backward compatibility
DEFAULT_PLATFORM_CODE = "oms"
class PlatformContextManager:
"""Manages platform context detection for multi-platform routing."""
@staticmethod
def detect_platform_context(request: Request) -> dict | None:
"""
Detect platform context from request.
Priority order:
1. Domain-based (production): oms.lu → platform code "oms"
2. Path-based (development): localhost:9999/oms/* → platform code "oms"
3. Default: localhost without path prefix → default platform
Returns dict with platform info or None if not detected.
"""
host = request.headers.get("host", "")
path = request.url.path
# Remove port from host if present (e.g., localhost:9999 -> localhost)
host_without_port = host.split(":")[0] if ":" in host else host
# Skip platform detection for admin routes - admin is global
if PlatformContextManager.is_admin_request(request):
return None
# Method 1: Domain-based detection (production)
# Check if the host matches a known platform domain
# This will be resolved in get_platform_from_context by DB lookup
if host_without_port and host_without_port not in ["localhost", "127.0.0.1"]:
# Could be a platform domain or a vendor subdomain/custom domain
# Check if it's a known platform domain pattern
# For now, assume non-localhost hosts that aren't subdomains are platform domains
if "." in host_without_port:
# This could be:
# - Platform domain: oms.lu, loyalty.lu
# - Vendor subdomain: vendor.oms.lu
# - Custom domain: shop.mycompany.com
# We detect platform domain vs subdomain by checking if it's a root domain
parts = host_without_port.split(".")
if len(parts) == 2: # e.g., oms.lu (root domain)
return {
"domain": host_without_port,
"detection_method": "domain",
"host": host,
"original_path": path,
}
# Method 2: Path-based detection (development)
# Check for path prefix like /oms/, /loyalty/
if path.startswith("/"):
path_parts = path[1:].split("/") # Remove leading / and split
if path_parts and path_parts[0]:
potential_platform_code = path_parts[0].lower()
# Check if this could be a platform code (not vendor paths)
if potential_platform_code not in [
"vendor",
"vendors",
"admin",
"api",
"static",
"media",
"assets",
"health",
"docs",
"redoc",
"openapi.json",
]:
return {
"path_prefix": potential_platform_code,
"detection_method": "path",
"host": host,
"original_path": path,
"clean_path": "/" + "/".join(path_parts[1:]) if len(path_parts) > 1 else "/",
}
# Method 3: Default platform for localhost without prefix
if host_without_port in ["localhost", "127.0.0.1"]:
return {
"path_prefix": DEFAULT_PLATFORM_CODE,
"detection_method": "default",
"host": host,
"original_path": path,
"clean_path": path,
}
return None
@staticmethod
def get_platform_from_context(db: Session, context: dict) -> Platform | None:
"""
Get platform from database using context information.
Supports:
1. Domain-based lookup (Platform.domain)
2. Path-prefix lookup (Platform.path_prefix)
3. Default lookup (Platform.code)
"""
if not context:
return None
platform = None
# Method 1: Domain-based lookup
if context.get("detection_method") == "domain":
domain = context.get("domain")
if domain:
platform = (
db.query(Platform)
.filter(Platform.domain == domain)
.filter(Platform.is_active == True)
.first()
)
if platform:
logger.debug(
f"[PLATFORM] Platform found via domain: {domain} → {platform.name}"
)
return platform
else:
logger.debug(f"[PLATFORM] No platform found for domain: {domain}")
# Method 2: Path-prefix lookup
if context.get("detection_method") == "path":
path_prefix = context.get("path_prefix")
if path_prefix:
# First try path_prefix, then try code
platform = (
db.query(Platform)
.filter(
(Platform.path_prefix == path_prefix) | (Platform.code == path_prefix)
)
.filter(Platform.is_active == True)
.first()
)
if platform:
logger.debug(
f"[PLATFORM] Platform found via path prefix: {path_prefix} → {platform.name}"
)
return platform
else:
logger.debug(f"[PLATFORM] No platform found for path prefix: {path_prefix}")
# Method 3: Default lookup
if context.get("detection_method") == "default":
platform = (
db.query(Platform)
.filter(Platform.code == DEFAULT_PLATFORM_CODE)
.filter(Platform.is_active == True)
.first()
)
if platform:
logger.debug(
f"[PLATFORM] Default platform found: {platform.name}"
)
return platform
return None
@staticmethod
def extract_clean_path(request: Request, platform_context: dict | None) -> str:
"""
Extract clean path without platform prefix for routing.
Downstream middleware (like VendorContextMiddleware) should use this
clean path for their detection logic.
"""
if not platform_context:
return request.url.path
# For path-based detection, use the pre-computed clean path
if platform_context.get("detection_method") == "path":
return platform_context.get("clean_path", request.url.path)
# For domain-based or default, path remains unchanged
return request.url.path
@staticmethod
def is_admin_request(request: Request) -> bool:
"""Check if request is for admin interface."""
host = request.headers.get("host", "")
path = request.url.path
if ":" in host:
host = host.split(":")[0]
if host.startswith("admin."):
return True
if path.startswith("/admin"):
return True
return False
@staticmethod
def is_static_file_request(request: Request) -> bool:
"""Check if request is for static files."""
path = request.url.path.lower()
static_extensions = (
".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg",
".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ".json",
".xml", ".txt", ".pdf", ".webmanifest",
)
static_paths = ("/static/", "/media/", "/assets/", "/.well-known/")
if path.endswith(static_extensions):
return True
if any(path.startswith(static_path) for static_path in static_paths):
return True
if "favicon.ico" in path:
return True
return False
class PlatformContextMiddleware(BaseHTTPMiddleware):
"""
Middleware to inject platform context into request state.
Runs BEFORE VendorContextMiddleware to establish platform context.
Sets:
request.state.platform: Platform object
request.state.platform_context: Detection metadata
request.state.platform_clean_path: Path without platform prefix
"""
async def dispatch(self, request: Request, call_next):
"""
Detect and inject platform context.
"""
# Skip platform detection for static files
if PlatformContextManager.is_static_file_request(request):
logger.debug(
f"[PLATFORM] Skipping platform detection for static file: {request.url.path}"
)
request.state.platform = None
request.state.platform_context = None
request.state.platform_clean_path = request.url.path
return await call_next(request)
# Skip platform detection for system endpoints
if request.url.path in ["/health", "/docs", "/redoc", "/openapi.json"]:
logger.debug(
f"[PLATFORM] Skipping platform detection for system path: {request.url.path}"
)
request.state.platform = None
request.state.platform_context = None
request.state.platform_clean_path = request.url.path
return await call_next(request)
# Admin requests are global (no platform context)
if PlatformContextManager.is_admin_request(request):
logger.debug(
f"[PLATFORM] Admin request - no platform context: {request.url.path}"
)
request.state.platform = None
request.state.platform_context = None
request.state.platform_clean_path = request.url.path
return await call_next(request)
# Detect platform context
platform_context = PlatformContextManager.detect_platform_context(request)
if platform_context:
db_gen = get_db()
db = next(db_gen)
try:
platform = PlatformContextManager.get_platform_from_context(
db, platform_context
)
if platform:
request.state.platform = platform
request.state.platform_context = platform_context
request.state.platform_clean_path = PlatformContextManager.extract_clean_path(
request, platform_context
)
logger.debug(
"[PLATFORM_CONTEXT] Platform detected",
extra={
"platform_id": platform.id,
"platform_code": platform.code,
"platform_name": platform.name,
"detection_method": platform_context.get("detection_method"),
"original_path": request.url.path,
"clean_path": request.state.platform_clean_path,
},
)
else:
# Platform code detected but not found in database
# This could be a vendor path like /vendors/...
logger.debug(
"[PLATFORM] Platform code not found, may be vendor path",
extra={
"context": platform_context,
"detection_method": platform_context.get("detection_method"),
},
)
request.state.platform = None
request.state.platform_context = None
request.state.platform_clean_path = request.url.path
finally:
db.close()
else:
logger.debug(
"[PLATFORM] No platform context detected",
extra={
"path": request.url.path,
"host": request.headers.get("host", ""),
},
)
request.state.platform = None
request.state.platform_context = None
request.state.platform_clean_path = request.url.path
# Continue to next middleware
return await call_next(request)
def get_current_platform(request: Request) -> Platform | None:
"""Helper function to get current platform from request state."""
return getattr(request.state, "platform", None)
def require_platform_context():
"""Dependency to require platform context in endpoints."""
def dependency(request: Request):
platform = get_current_platform(request)
if not platform:
from fastapi import HTTPException
raise HTTPException(
status_code=404,
detail="Platform not found"
)
return platform
return dependency

5
models/database/__init__.py
View File

@@ -17,6 +17,8 @@ from .architecture_scan import (
from .base import Base
from .company import Company
from .content_page import ContentPage
from .platform import Platform
from .vendor_platform import VendorPlatform
from .customer import Customer, CustomerAddress
from .password_reset_token import PasswordResetToken
from .email import EmailCategory, EmailLog, EmailStatus, EmailTemplate
@@ -106,6 +108,9 @@ __all__ = [
"VendorTheme",
# Content
"ContentPage",
# Platform
"Platform",
"VendorPlatform",
# Customer & Auth
"Customer",
"CustomerAddress",

103
models/database/content_page.py
View File

@@ -3,15 +3,27 @@
Content Page Model
Manages static content pages (About, FAQ, Contact, Shipping, Returns, etc.)
with platform-level defaults and vendor-specific overrides.
with a three-tier hierarchy:
1. Platform Marketing Pages (is_platform_page=True, vendor_id=NULL)
- Homepage, pricing, platform about, contact
- Describes the platform/business offering itself
2. Vendor Default Pages (is_platform_page=False, vendor_id=NULL)
- Generic storefront pages that all vendors inherit
- 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
Features:
- Platform-level default content
- Vendor-specific overrides
- Multi-platform support (each platform has its own pages)
- Three-tier content resolution
- Rich text content (HTML/Markdown)
- SEO metadata
- Published/Draft status
- Version history support
- Navigation placement (header, footer, legal)
"""
from datetime import UTC, datetime
@@ -34,25 +46,50 @@ from app.core.database import Base
class ContentPage(Base):
"""
Content pages for shops (About, FAQ, Contact, etc.)
Content pages with three-tier hierarchy.
Two-tier system:
1. Platform-level defaults (vendor_id=NULL)
2. Vendor-specific overrides (vendor_id=123)
Page Types:
1. Platform Marketing Page: platform_id=X, vendor_id=NULL, is_platform_page=True
- Platform's own pages (homepage, pricing, about)
2. Vendor Default Page: platform_id=X, vendor_id=NULL, is_platform_page=False
- Fallback pages for vendors who haven't customized
3. Vendor Override/Custom: platform_id=X, vendor_id=Y, is_platform_page=False
- Vendor-specific content
Lookup logic:
1. Check for vendor-specific page (vendor_id + slug)
2. If not found, use platform default (slug only)
3. If neither exists, show 404 or default template
Resolution Logic:
1. Check for vendor override (platform_id + vendor_id + slug)
2. Fall back to vendor default (platform_id + vendor_id=NULL + is_platform_page=False)
3. If neither exists, return 404
"""
__tablename__ = "content_pages"
id = Column(Integer, primary_key=True, index=True)
# Vendor association (NULL = platform default)
# Platform association (REQUIRED - determines which platform this page belongs to)
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="Platform this page belongs to",
)
# Vendor association (NULL = platform page or vendor default)
vendor_id = Column(
Integer, ForeignKey("vendors.id", ondelete="CASCADE"), nullable=True, index=True
Integer,
ForeignKey("vendors.id", ondelete="CASCADE"),
nullable=True,
index=True,
comment="Vendor this page belongs to (NULL for platform/default pages)",
)
# Distinguish platform marketing pages from vendor defaults
is_platform_page = Column(
Boolean,
default=False,
nullable=False,
comment="True = platform marketing page (homepage, pricing); False = vendor default or override",
)
# Page identification
@@ -106,18 +143,22 @@ class ContentPage(Base):
)
# Relationships
platform = relationship("Platform", back_populates="content_pages")
vendor = relationship("Vendor", back_populates="content_pages")
creator = relationship("User", foreign_keys=[created_by])
updater = relationship("User", foreign_keys=[updated_by])
# Constraints
__table_args__ = (
# Unique combination: vendor can only have one page per slug
# Platform defaults (vendor_id=NULL) can only have one page per slug
UniqueConstraint("vendor_id", "slug", name="uq_vendor_slug"),
# Unique combination: platform + vendor + slug
# Platform pages: platform_id + vendor_id=NULL + is_platform_page=True
# Vendor defaults: platform_id + vendor_id=NULL + is_platform_page=False
# Vendor overrides: platform_id + vendor_id + slug
UniqueConstraint("platform_id", "vendor_id", "slug", name="uq_platform_vendor_slug"),
# Indexes for performance
Index("idx_vendor_published", "vendor_id", "is_published"),
Index("idx_slug_published", "slug", "is_published"),
Index("idx_platform_vendor_published", "platform_id", "vendor_id", "is_published"),
Index("idx_platform_slug_published", "platform_id", "slug", "is_published"),
Index("idx_platform_page_type", "platform_id", "is_platform_page"),
)
def __repr__(self):
@@ -125,19 +166,31 @@ class ContentPage(Base):
return f"<ContentPage(id={self.id}, vendor={vendor_name}, slug={self.slug}, title={self.title})>"
@property
def is_platform_default(self):
"""Check if this is a platform-level default page."""
return self.vendor_id is None
def is_vendor_default(self):
"""Check if this is a vendor default page (fallback for all vendors)."""
return self.vendor_id is None and not self.is_platform_page
@property
def is_vendor_override(self):
"""Check if this is a vendor-specific override."""
"""Check if this is a vendor-specific override or custom page."""
return self.vendor_id is not None
@property
def page_tier(self) -> str:
"""Get the tier level of this page for display purposes."""
if self.is_platform_page:
return "platform"
elif self.vendor_id is None:
return "vendor_default"
else:
return "vendor_override"
def to_dict(self):
"""Convert to dictionary for API responses."""
return {
"id": self.id,
"platform_id": self.platform_id,
"platform_code": self.platform.code if self.platform else None,
"vendor_id": self.vendor_id,
"vendor_name": self.vendor.name if self.vendor else None,
"slug": self.slug,
@@ -155,8 +208,10 @@ class ContentPage(Base):
"show_in_footer": self.show_in_footer,
"show_in_header": self.show_in_header,
"show_in_legal": self.show_in_legal,
"is_platform_default": self.is_platform_default,
"is_platform_page": self.is_platform_page,
"is_vendor_default": self.is_vendor_default,
"is_vendor_override": self.is_vendor_override,
"page_tier": self.page_tier,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
"created_by": self.created_by,

9
models/database/feature.py
View File

@@ -33,6 +33,7 @@ class FeatureCategory(str, enum.Enum):
TEAM = "team"
BRANDING = "branding"
CUSTOMERS = "customers"
CMS = "cms"
class FeatureUILocation(str, enum.Enum):
@@ -189,3 +190,11 @@ class FeatureCode:
CUSTOMER_VIEW = "customer_view"
CUSTOMER_EXPORT = "customer_export"
CUSTOMER_MESSAGING = "customer_messaging"
# CMS
CMS_BASIC = "cms_basic" # Basic CMS functionality (override defaults)
CMS_CUSTOM_PAGES = "cms_custom_pages" # Create custom pages beyond defaults
CMS_UNLIMITED_PAGES = "cms_unlimited_pages" # No page limit
CMS_TEMPLATES = "cms_templates" # Access to page templates
CMS_SEO = "cms_seo" # Advanced SEO features
CMS_SCHEDULING = "cms_scheduling" # Schedule page publish/unpublish

218
models/database/platform.py Normal file
View File

@@ -0,0 +1,218 @@
# models/database/platform.py
"""
Platform model representing a business offering/product line.
Platforms are independent business products (e.g., OMS, Loyalty Program, Site Builder)
that can have their own:
- Marketing pages (homepage, pricing, about)
- Vendor default pages (fallback storefront pages)
- Subscription tiers with platform-specific features
- Branding and configuration
Each vendor can belong to multiple platforms via the VendorPlatform junction table.
"""
from sqlalchemy import (
JSON,
Boolean,
Column,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class Platform(Base, TimestampMixin):
"""
Represents a business offering/product line.
Examples:
- Wizamart OMS (Order Management System)
- Loyalty+ (Loyalty Program Platform)
- Site Builder (Website Builder for Local Businesses)
Each platform has:
- Its own domain (production) or path prefix (development)
- Independent CMS pages (marketing pages + vendor defaults)
- Platform-specific subscription tiers
- Custom branding and theme
"""
__tablename__ = "platforms"
id = Column(Integer, primary_key=True, index=True)
# ========================================================================
# Identity
# ========================================================================
code = Column(
String(50),
unique=True,
nullable=False,
index=True,
comment="Unique platform identifier (e.g., 'oms', 'loyalty', 'sites')",
)
name = Column(
String(100),
nullable=False,
comment="Display name (e.g., 'Wizamart OMS')",
)
description = Column(
Text,
nullable=True,
comment="Platform description for admin/marketing purposes",
)
# ========================================================================
# Domain Routing
# ========================================================================
domain = Column(
String(255),
unique=True,
nullable=True,
index=True,
comment="Production domain (e.g., 'oms.lu', 'loyalty.lu')",
)
path_prefix = Column(
String(50),
unique=True,
nullable=True,
index=True,
comment="Development path prefix (e.g., 'oms' for localhost:9999/oms/*)",
)
# ========================================================================
# Branding
# ========================================================================
logo = Column(
String(500),
nullable=True,
comment="Logo URL for light mode",
)
logo_dark = Column(
String(500),
nullable=True,
comment="Logo URL for dark mode",
)
favicon = Column(
String(500),
nullable=True,
comment="Favicon URL",
)
theme_config = Column(
JSON,
nullable=True,
default=dict,
comment="Theme configuration (colors, fonts, etc.)",
)
# ========================================================================
# Localization
# ========================================================================
default_language = Column(
String(5),
default="fr",
nullable=False,
comment="Default language code (e.g., 'fr', 'en', 'de')",
)
supported_languages = Column(
JSON,
default=["fr", "de", "en"],
nullable=False,
comment="List of supported language codes",
)
# ========================================================================
# Status
# ========================================================================
is_active = Column(
Boolean,
default=True,
nullable=False,
comment="Whether the platform is active and accessible",
)
is_public = Column(
Boolean,
default=True,
nullable=False,
comment="Whether the platform is visible in public listings",
)
# ========================================================================
# Configuration
# ========================================================================
settings = Column(
JSON,
nullable=True,
default=dict,
comment="Platform-specific settings and feature flags",
)
# ========================================================================
# Relationships
# ========================================================================
# Content pages belonging to this platform
content_pages = relationship(
"ContentPage",
back_populates="platform",
cascade="all, delete-orphan",
)
# Vendors on this platform (via junction table)
vendor_platforms = relationship(
"VendorPlatform",
back_populates="platform",
cascade="all, delete-orphan",
)
# Subscription tiers for this platform
subscription_tiers = relationship(
"SubscriptionTier",
back_populates="platform",
foreign_keys="SubscriptionTier.platform_id",
)
# ========================================================================
# Indexes
# ========================================================================
__table_args__ = (
Index("idx_platform_active", "is_active"),
Index("idx_platform_public", "is_public", "is_active"),
)
# ========================================================================
# Properties
# ========================================================================
@property
def base_url(self) -> str:
"""Get the base URL for this platform (for link generation)."""
if self.domain:
return f"https://{self.domain}"
if self.path_prefix:
return f"/{self.path_prefix}"
return "/"
def __repr__(self) -> str:
return f"<Platform(code='{self.code}', name='{self.name}')>"

45
models/database/subscription.py
View File

@@ -84,12 +84,26 @@ class SubscriptionTier(Base, TimestampMixin):
Database-driven tier definitions with Stripe integration.
Replaces the hardcoded TIER_LIMITS dict for dynamic tier management.
Can be:
- Global tier (platform_id=NULL): Available to all platforms
- Platform-specific tier (platform_id set): Only for that platform
"""
__tablename__ = "subscription_tiers"
id = Column(Integer, primary_key=True, index=True)
code = Column(String(30), unique=True, nullable=False, index=True)
# Platform association (NULL = global tier available to all platforms)
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=True,
index=True,
comment="Platform this tier belongs to (NULL = global tier)",
)
code = Column(String(30), nullable=False, index=True)
name = Column(String(100), nullable=False)
description = Column(Text, nullable=True)
@@ -103,6 +117,18 @@ class SubscriptionTier(Base, TimestampMixin):
team_members = Column(Integer, nullable=True)
order_history_months = Column(Integer, nullable=True)
# CMS Limits (null = unlimited)
cms_pages_limit = Column(
Integer,
nullable=True,
comment="Total CMS pages limit (NULL = unlimited)",
)
cms_custom_pages_limit = Column(
Integer,
nullable=True,
comment="Custom pages limit, excluding overrides (NULL = unlimited)",
)
# Features (JSON array of feature codes)
features = Column(JSON, default=list)
@@ -116,8 +142,21 @@ class SubscriptionTier(Base, TimestampMixin):
is_active = Column(Boolean, default=True, nullable=False)
is_public = Column(Boolean, default=True, nullable=False) # False for enterprise
# Relationship to Platform
platform = relationship(
"Platform",
back_populates="subscription_tiers",
foreign_keys=[platform_id],
)
# Unique constraint: tier code must be unique per platform (or globally if NULL)
__table_args__ = (
Index("idx_tier_platform_active", "platform_id", "is_active"),
)
def __repr__(self):
return f"<SubscriptionTier(code='{self.code}', name='{self.name}')>"
platform_info = f", platform_id={self.platform_id}" if self.platform_id else ""
return f"<SubscriptionTier(code='{self.code}', name='{self.name}'{platform_info})>"
def to_dict(self) -> dict:
"""Convert tier to dictionary (compatible with TIER_LIMITS format)."""
@@ -129,6 +168,8 @@ class SubscriptionTier(Base, TimestampMixin):
"products_limit": self.products_limit,
"team_members": self.team_members,
"order_history_months": self.order_history_months,
"cms_pages_limit": self.cms_pages_limit,
"cms_custom_pages_limit": self.cms_custom_pages_limit,
"features": self.features or [],
}

7
models/database/vendor.py
View File

@@ -245,6 +245,13 @@ class Vendor(Base, TimestampMixin):
cascade="all, delete-orphan",
)
# Platform memberships (many-to-many via junction table)
vendor_platforms = relationship(
"VendorPlatform",
back_populates="vendor",
cascade="all, delete-orphan",
)
def __repr__(self):
"""String representation of the Vendor object."""
return f"<Vendor(id={self.id}, vendor_code='{self.vendor_code}', name='{self.name}', subdomain='{self.subdomain}')>"

189
models/database/vendor_platform.py Normal file
View File

@@ -0,0 +1,189 @@
# models/database/vendor_platform.py
"""
VendorPlatform junction table for many-to-many relationship between Vendor and Platform.
A vendor CAN belong to multiple platforms (e.g., both OMS and Loyalty Program).
Each membership can have:
- Platform-specific subscription tier
- Custom subdomain for that platform
- Platform-specific settings
- Active/inactive status
"""
from datetime import UTC, datetime
from sqlalchemy import (
JSON,
Boolean,
Column,
DateTime,
ForeignKey,
Index,
Integer,
String,
UniqueConstraint,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class VendorPlatform(Base, TimestampMixin):
"""
Junction table linking vendors to platforms.
Allows a vendor to:
- Subscribe to multiple platforms (OMS + Loyalty)
- Have different tiers per platform
- Have platform-specific subdomains
- Store platform-specific settings
Example:
- Vendor "WizaMart" is on OMS platform (Professional tier)
- Vendor "WizaMart" is also on Loyalty platform (Basic tier)
"""
__tablename__ = "vendor_platforms"
id = Column(Integer, primary_key=True, index=True)
# ========================================================================
# Foreign Keys
# ========================================================================
vendor_id = Column(
Integer,
ForeignKey("vendors.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="Reference to the vendor",
)
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="Reference to the platform",
)
tier_id = Column(
Integer,
ForeignKey("subscription_tiers.id", ondelete="SET NULL"),
nullable=True,
index=True,
comment="Platform-specific subscription tier",
)
# ========================================================================
# Membership Status
# ========================================================================
is_active = Column(
Boolean,
default=True,
nullable=False,
comment="Whether the vendor is active on this platform",
)
is_primary = Column(
Boolean,
default=False,
nullable=False,
comment="Whether this is the vendor's primary platform",
)
# ========================================================================
# Platform-Specific Configuration
# ========================================================================
custom_subdomain = Column(
String(100),
nullable=True,
comment="Platform-specific subdomain (if different from main subdomain)",
)
settings = Column(
JSON,
nullable=True,
default=dict,
comment="Platform-specific vendor settings",
)
# ========================================================================
# Timestamps
# ========================================================================
joined_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
nullable=False,
comment="When the vendor joined this platform",
)
# ========================================================================
# Relationships
# ========================================================================
vendor = relationship(
"Vendor",
back_populates="vendor_platforms",
)
platform = relationship(
"Platform",
back_populates="vendor_platforms",
)
tier = relationship(
"SubscriptionTier",
foreign_keys=[tier_id],
)
# ========================================================================
# Constraints & Indexes
# ========================================================================
__table_args__ = (
# Each vendor can only be on a platform once
UniqueConstraint(
"vendor_id",
"platform_id",
name="uq_vendor_platform",
),
# Performance indexes
Index(
"idx_vendor_platform_active",
"vendor_id",
"platform_id",
"is_active",
),
Index(
"idx_vendor_platform_primary",
"vendor_id",
"is_primary",
),
)
# ========================================================================
# Properties
# ========================================================================
@property
def tier_code(self) -> str | None:
"""Get the tier code for this platform membership."""
return self.tier.code if self.tier else None
@property
def tier_name(self) -> str | None:
"""Get the tier name for this platform membership."""
return self.tier.name if self.tier else None
def __repr__(self) -> str:
return (
f"<VendorPlatform("
f"vendor_id={self.vendor_id}, "
f"platform_id={self.platform_id}, "
f"is_active={self.is_active})>"
)