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

refactor: complete module-driven architecture migration

Browse Source 
This commit completes the migration to a fully module-driven architecture:

## Models Migration
- Moved all domain models from models/database/ to their respective modules:
  - tenancy: User, Admin, Vendor, Company, Platform, VendorDomain, etc.
  - cms: MediaFile, VendorTheme
  - messaging: Email, VendorEmailSettings, VendorEmailTemplate
  - core: AdminMenuConfig
- models/database/ now only contains Base and TimestampMixin (infrastructure)

## Schemas Migration
- Moved all domain schemas from models/schema/ to their respective modules:
  - tenancy: company, vendor, admin, team, vendor_domain
  - cms: media, image, vendor_theme
  - messaging: email
- models/schema/ now only contains base.py and auth.py (infrastructure)

## Routes Migration
- Moved admin routes from app/api/v1/admin/ to modules:
  - menu_config.py -> core module
  - modules.py -> tenancy module
  - module_config.py -> tenancy module
- app/api/v1/admin/ now only aggregates auto-discovered module routes

## Menu System
- Implemented module-driven menu system with MenuDiscoveryService
- Extended FrontendType enum: PLATFORM, ADMIN, VENDOR, STOREFRONT
- Added MenuItemDefinition and MenuSectionDefinition dataclasses
- Each module now defines its own menu items in definition.py
- MenuService integrates with MenuDiscoveryService for template rendering

## Documentation
- Updated docs/architecture/models-structure.md
- Updated docs/architecture/menu-management.md
- Updated architecture validation rules for new exceptions

## Architecture Validation
- Updated MOD-019 rule to allow base.py in models/schema/
- Created core module exceptions.py and schemas/ directory
- All validation errors resolved (only warnings remain)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2026-02-01 21:02:56 +01:00
parent 09d7d282c6
commit d7a0ff8818
307 changed files with 5536 additions and 3826 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
models/__init__.py
View File

@@ -4,24 +4,20 @@
# API models (Pydantic) - import the modules, not all classes
from . import schema
# Database models (SQLAlchemy)
# Database models (SQLAlchemy) - base only, avoid circular imports
from .database.base import Base
from .database.user import User
from .database.vendor import Vendor
# Module-based models
from app.modules.inventory.models import Inventory
from app.modules.marketplace.models import MarketplaceImportJob, MarketplaceProduct
from app.modules.catalog.models import Product
# Note: User, Vendor, and other domain models should be imported from
# their canonical locations in app/modules/*/models/
#
# Example:
# from app.modules.tenancy.models import User, Vendor
# from app.modules.catalog.models import Product
# from app.modules.inventory.models import Inventory
# from app.modules.marketplace.models import MarketplaceImportJob, MarketplaceProduct
# Export database models for Alembic
# Export database base for Alembic
__all__ = [
"Base",
"User",
"MarketplaceProduct",
"Inventory",
"Vendor",
"Product",
"MarketplaceImportJob",
"api", # API models namespace
"schema", # API models namespace
]

266
models/database/__init__.py
View File

@@ -1,256 +1,30 @@
# models/database/__init__.py
"""
Database models package.
Database models package - Base classes and mixins only.
This package imports all SQLAlchemy models to ensure they are registered
with Base.metadata. This includes:
1. Core models (defined in this directory)
2. Module models (discovered from app/modules/<module>/models/)
This package provides the base infrastructure for SQLAlchemy models:
- Base: SQLAlchemy declarative base
- TimestampMixin: Mixin for created_at/updated_at timestamps
Module Model Discovery:
- Modules can define their own models in app/modules/<module>/models/
- These are automatically imported when this package loads
- Module models must use `from app.core.database import Base`
IMPORTANT: Domain models have been migrated to their respective modules:
- Tenancy models: app.modules.tenancy.models
- Catalog models: app.modules.catalog.models
- Orders models: app.modules.orders.models
- Inventory models: app.modules.inventory.models
- Billing models: app.modules.billing.models
- Messaging models: app.modules.messaging.models
- CMS models: app.modules.cms.models
- Marketplace models: app.modules.marketplace.models
- Customers models: app.modules.customers.models
- Dev Tools models: app.modules.dev_tools.models
- Core models: app.modules.core.models
Import models from their canonical module locations instead of this package.
"""
import importlib
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
# ============================================================================
# CORE MODELS (always loaded)
# ============================================================================
from .admin import (
AdminAuditLog,
AdminSession,
AdminSetting,
PlatformAlert,
)
from app.modules.messaging.models import AdminNotification
from .admin_menu_config import AdminMenuConfig, FrontendType, MANDATORY_MENU_ITEMS
from .admin_platform import AdminPlatform
from app.modules.dev_tools.models import (
ArchitectureScan,
ArchitectureViolation,
ViolationAssignment,
ViolationComment,
)
from .base import Base
from .company import Company
from .platform import Platform
from .platform_module import PlatformModule
from .vendor_platform import VendorPlatform
from app.modules.customers.models import Customer, CustomerAddress
from app.modules.customers.models import PasswordResetToken
from .email import EmailCategory, EmailLog, EmailStatus, EmailTemplate
from .vendor_email_template import VendorEmailTemplate
from .vendor_email_settings import EmailProvider, VendorEmailSettings, PREMIUM_EMAIL_PROVIDERS
from app.modules.billing.models import Feature, FeatureCategory, FeatureCode, FeatureUILocation
from app.modules.inventory.models import Inventory
from app.modules.inventory.models import InventoryTransaction, TransactionType
from app.modules.orders.models import (
Invoice,
InvoiceStatus,
VATRegime,
VendorInvoiceSettings,
)
from app.modules.marketplace.models import (
LetzshopFulfillmentQueue,
LetzshopHistoricalImportJob,
LetzshopSyncLog,
VendorLetzshopCredentials,
MarketplaceImportError,
MarketplaceImportJob,
DigitalDeliveryMethod,
MarketplaceProduct,
ProductType,
MarketplaceProductTranslation,
)
from app.modules.messaging.models import (
Conversation,
ConversationParticipant,
ConversationType,
Message,
MessageAttachment,
ParticipantType,
)
from .media import MediaFile
from app.modules.catalog.models import ProductMedia
from app.modules.marketplace.models import OnboardingStatus, OnboardingStep, VendorOnboarding
from app.modules.orders.models import Order, OrderItem
from app.modules.orders.models import OrderItemException
from app.modules.catalog.models import Product, ProductTranslation
from app.modules.billing.models import (
AddOnCategory,
AddOnProduct,
BillingHistory,
BillingPeriod,
StripeWebhookEvent,
SubscriptionStatus,
SubscriptionTier,
TierCode,
TIER_LIMITS,
VendorAddOn,
VendorSubscription,
)
from app.modules.dev_tools.models import TestCollection, TestResult, TestRun
from .user import User
from .vendor import Role, Vendor, VendorUser
from .vendor_domain import VendorDomain
from .vendor_theme import VendorTheme
# ============================================================================
# MODULE MODELS (dynamically discovered)
# ============================================================================
def _discover_module_models():
"""
Discover and import models from app/modules/<module>/models/ directories.
This ensures module models are registered with Base.metadata for:
1. Alembic migrations
2. SQLAlchemy queries
Module models must:
- Be in app/modules/<module>/models/__init__.py or individual files
- Import Base from app.core.database
"""
modules_dir = Path(__file__).parent.parent.parent / "app" / "modules"
if not modules_dir.exists():
return
for module_dir in sorted(modules_dir.iterdir()):
if not module_dir.is_dir():
continue
models_init = module_dir / "models" / "__init__.py"
if models_init.exists():
module_name = f"app.modules.{module_dir.name}.models"
try:
importlib.import_module(module_name)
logger.debug(f"[Models] Loaded module models: {module_name}")
except ImportError as e:
logger.warning(f"[Models] Failed to import {module_name}: {e}")
# Run discovery at import time
_discover_module_models()
# ============================================================================
# EXPORTS
# ============================================================================
from .base import Base, TimestampMixin
__all__ = [
# Admin-specific models
"AdminAuditLog",
"AdminMenuConfig",
"FrontendType",
"AdminNotification",
"AdminPlatform",
"AdminSetting",
"MANDATORY_MENU_ITEMS",
"PlatformAlert",
"AdminSession",
# Architecture/Code Quality
"ArchitectureScan",
"ArchitectureViolation",
"ViolationAssignment",
"ViolationComment",
# Test Runs
"TestRun",
"TestResult",
"TestCollection",
# Base
"Base",
# User & Auth
"User",
# Company & Vendor
"Company",
"Vendor",
"VendorUser",
"Role",
"VendorDomain",
"VendorTheme",
# Platform
"Platform",
"PlatformModule",
"VendorPlatform",
# Customer & Auth
"Customer",
"CustomerAddress",
"PasswordResetToken",
# Email
"EmailCategory",
"EmailLog",
"EmailStatus",
"EmailTemplate",
"VendorEmailTemplate",
"VendorEmailSettings",
"EmailProvider",
"PREMIUM_EMAIL_PROVIDERS",
# Features
"Feature",
"FeatureCategory",
"FeatureCode",
"FeatureUILocation",
# Product - Enums
"ProductType",
"DigitalDeliveryMethod",
# Product - Models
"MarketplaceProduct",
"MarketplaceProductTranslation",
"Product",
"ProductTranslation",
# Import
"MarketplaceImportJob",
"MarketplaceImportError",
# Inventory
"Inventory",
"InventoryTransaction",
"TransactionType",
# Media
"MediaFile",
"ProductMedia",
# Invoicing
"Invoice",
"InvoiceStatus",
"VATRegime",
"VendorInvoiceSettings",
# Orders
"Order",
"OrderItem",
"OrderItemException",
# Letzshop Integration
"VendorLetzshopCredentials",
"LetzshopFulfillmentQueue",
"LetzshopSyncLog",
"LetzshopHistoricalImportJob",
# Subscription & Billing
"VendorSubscription",
"SubscriptionStatus",
"SubscriptionTier",
"TierCode",
"TIER_LIMITS",
"AddOnProduct",
"AddOnCategory",
"BillingPeriod",
"VendorAddOn",
"BillingHistory",
"StripeWebhookEvent",
# Messaging
"Conversation",
"ConversationParticipant",
"ConversationType",
"Message",
"MessageAttachment",
"ParticipantType",
# Onboarding
"OnboardingStatus",
"OnboardingStep",
"VendorOnboarding",
"TimestampMixin",
]

198
models/database/admin.py
View File

@@ -1,198 +0,0 @@
# Admin-specific models
# models/database/admin.py
"""
Admin-specific database models.
This module provides models for:
- Admin audit logging (compliance and security tracking)
- Admin notifications (system alerts and warnings)
- Platform settings (global configuration)
- Platform alerts (system-wide issues)
- Application logs (critical events logging)
"""
from sqlalchemy import (
JSON,
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from .base import TimestampMixin
class AdminAuditLog(Base, TimestampMixin):
"""
Track all admin actions for compliance and security.
Separate from regular audit logs - focuses on admin-specific operations
like vendor creation, user management, and system configuration changes.
"""
__tablename__ = "admin_audit_logs"
id = Column(Integer, primary_key=True, index=True)
admin_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
action = Column(
String(100), nullable=False, index=True
) # create_vendor, delete_vendor, etc.
target_type = Column(
String(50), nullable=False, index=True
) # vendor, user, import_job, setting
target_id = Column(String(100), nullable=False, index=True)
details = Column(JSON) # Additional context about the action
ip_address = Column(String(45)) # IPv4 or IPv6
user_agent = Column(Text)
request_id = Column(String(100)) # For correlating with application logs
# Relationships
admin_user = relationship("User", foreign_keys=[admin_user_id])
def __repr__(self):
return f"<AdminAuditLog(id={self.id}, action='{self.action}', target={self.target_type}:{self.target_id})>"
# AdminNotification has been moved to app/modules/messaging/models/admin_notification.py
# It's re-exported via models/database/__init__.py for backwards compatibility
class AdminSetting(Base, TimestampMixin):
"""
Platform-wide admin settings and configuration.
Stores global settings that affect the entire platform, different from
vendor-specific settings. Supports encryption for sensitive values.
Examples:
- max_vendors_allowed
- maintenance_mode
- default_vendor_trial_days
- smtp_settings
- stripe_api_keys (encrypted)
"""
__tablename__ = "admin_settings"
id = Column(Integer, primary_key=True, index=True)
key = Column(String(100), unique=True, nullable=False, index=True)
value = Column(Text, nullable=False)
value_type = Column(String(20), default="string") # string, integer, boolean, json
category = Column(
String(50), index=True
) # system, security, marketplace, notifications
description = Column(Text)
is_encrypted = Column(Boolean, default=False)
is_public = Column(Boolean, default=False) # Can be exposed to frontend?
last_modified_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
# Relationships
last_modified_by = relationship("User", foreign_keys=[last_modified_by_user_id])
def __repr__(self):
return f"<AdminSetting(key='{self.key}', category='{self.category}')>"
class PlatformAlert(Base, TimestampMixin):
"""
System-wide alerts that admins need to be aware of.
Tracks platform issues, performance problems, security incidents,
and other system-level concerns that require admin attention.
"""
__tablename__ = "platform_alerts"
id = Column(Integer, primary_key=True, index=True)
alert_type = Column(
String(50), nullable=False, index=True
) # security, performance, capacity, integration
severity = Column(
String(20), nullable=False, index=True
) # info, warning, error, critical
title = Column(String(200), nullable=False)
description = Column(Text)
affected_vendors = Column(JSON) # List of affected vendor IDs
affected_systems = Column(JSON) # List of affected system components
is_resolved = Column(Boolean, default=False, index=True)
resolved_at = Column(DateTime, nullable=True)
resolved_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
resolution_notes = Column(Text)
auto_generated = Column(Boolean, default=True) # System-generated vs manual
occurrence_count = Column(Integer, default=1) # Track repeated occurrences
first_occurred_at = Column(DateTime, nullable=False)
last_occurred_at = Column(DateTime, nullable=False)
# Relationships
resolved_by = relationship("User", foreign_keys=[resolved_by_user_id])
def __repr__(self):
return f"<PlatformAlert(id={self.id}, type='{self.alert_type}', severity='{self.severity}')>"
class AdminSession(Base, TimestampMixin):
"""
Track admin login sessions for security monitoring.
Helps identify suspicious login patterns, track concurrent sessions,
and enforce session policies for admin users.
"""
__tablename__ = "admin_sessions"
id = Column(Integer, primary_key=True, index=True)
admin_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
session_token = Column(String(255), unique=True, nullable=False, index=True)
ip_address = Column(String(45), nullable=False)
user_agent = Column(Text)
login_at = Column(DateTime, nullable=False, index=True)
last_activity_at = Column(DateTime, nullable=False)
logout_at = Column(DateTime, nullable=True)
is_active = Column(Boolean, default=True, index=True)
logout_reason = Column(String(50)) # manual, timeout, forced, suspicious
# Relationships
admin_user = relationship("User", foreign_keys=[admin_user_id])
def __repr__(self):
return f"<AdminSession(id={self.id}, admin_user_id={self.admin_user_id}, is_active={self.is_active})>"
class ApplicationLog(Base, TimestampMixin):
"""
Application-level logs stored in database for critical events.
Stores WARNING, ERROR, and CRITICAL level logs for easy searching,
filtering, and compliance. INFO and DEBUG logs are kept in files only.
"""
__tablename__ = "application_logs"
id = Column(Integer, primary_key=True, index=True)
timestamp = Column(DateTime, nullable=False, index=True)
level = Column(String(20), nullable=False, index=True) # WARNING, ERROR, CRITICAL
logger_name = Column(String(200), nullable=False, index=True)
module = Column(String(200))
function_name = Column(String(100))
line_number = Column(Integer)
message = Column(Text, nullable=False)
exception_type = Column(String(200))
exception_message = Column(Text)
stack_trace = Column(Text)
request_id = Column(String(100), index=True) # For correlating logs
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=True, index=True)
context = Column(JSON) # Additional context data
# Relationships
user = relationship("User", foreign_keys=[user_id])
vendor = relationship("Vendor", foreign_keys=[vendor_id])
def __repr__(self):
return f"<ApplicationLog(id={self.id}, level='{self.level}', logger='{self.logger_name}')>"

223
models/database/admin_menu_config.py
View File

@@ -1,223 +0,0 @@
# models/database/admin_menu_config.py
"""
Menu visibility configuration for admin and vendor frontends.
Supports two frontend types:
- 'admin': Admin panel menus (for super admins and platform admins)
- 'vendor': Vendor dashboard menus (configured per platform)
Supports two scopes:
- Platform-level: Menu config for a platform (platform_id is set)
→ For admin frontend: applies to platform admins
→ For vendor frontend: applies to all vendors on that platform
- User-level: Menu config for a specific super admin (user_id is set)
→ Only for admin frontend (super admins configuring their own menu)
Design:
- Opt-out model: All items visible by default, store hidden items
- Mandatory items: Some items cannot be hidden (defined per frontend type)
- Only stores non-default state (is_visible=False) to keep table small
"""
from sqlalchemy import (
Boolean,
CheckConstraint,
Column,
Enum,
ForeignKey,
Index,
Integer,
String,
UniqueConstraint,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
# Import FrontendType and MANDATORY_MENU_ITEMS from the central location
# and re-export for backward compatibility with existing imports.
# These were moved to app.modules.enums to break a circular import:
# app.modules.base -> models.database -> model discovery -> module definitions -> app.modules.base
from app.modules.enums import FrontendType, MANDATORY_MENU_ITEMS
class AdminMenuConfig(Base, TimestampMixin):
"""
Menu visibility configuration for admin and vendor frontends.
Supports two frontend types:
- 'admin': Admin panel menus
- 'vendor': Vendor dashboard menus
Supports two scopes:
- Platform scope: platform_id is set
→ Admin: applies to platform admins of that platform
→ Vendor: applies to all vendors on that platform
- User scope: user_id is set (admin frontend only)
→ Applies to a specific super admin user
Resolution order for admin frontend:
- Platform admins: Check platform config → fall back to default
- Super admins: Check user config → fall back to default
Resolution order for vendor frontend:
- Check platform config → fall back to default
Examples:
- Platform "OMS" wants to hide "inventory" from admin panel
→ frontend_type='admin', platform_id=1, menu_item_id="inventory", is_visible=False
- Platform "OMS" wants to hide "letzshop" from vendor dashboard
→ frontend_type='vendor', platform_id=1, menu_item_id="letzshop", is_visible=False
- Super admin "john" wants to hide "code-quality" from their admin panel
→ frontend_type='admin', user_id=5, menu_item_id="code-quality", is_visible=False
"""
__tablename__ = "admin_menu_configs"
id = Column(Integer, primary_key=True, index=True)
# ========================================================================
# Frontend Type
# ========================================================================
frontend_type = Column(
Enum(FrontendType, values_callable=lambda obj: [e.value for e in obj]),
nullable=False,
default=FrontendType.ADMIN,
index=True,
comment="Which frontend this config applies to (admin or vendor)",
)
# ========================================================================
# Scope: Platform scope OR User scope (for admin frontend only)
# ========================================================================
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=True,
index=True,
comment="Platform scope - applies to users/vendors of this platform",
)
user_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=True,
index=True,
comment="User scope - applies to this specific super admin (admin frontend only)",
)
# ========================================================================
# Menu Item Configuration
# ========================================================================
menu_item_id = Column(
String(50),
nullable=False,
index=True,
comment="Menu item identifier from registry (e.g., 'products', 'inventory')",
)
is_visible = Column(
Boolean,
default=True,
nullable=False,
comment="Whether this menu item is visible (False = hidden)",
)
# ========================================================================
# Relationships
# ========================================================================
platform = relationship(
"Platform",
back_populates="menu_configs",
)
user = relationship(
"User",
back_populates="menu_configs",
)
# ========================================================================
# Constraints
# ========================================================================
__table_args__ = (
# Unique constraint: one config per frontend+platform+menu_item
UniqueConstraint(
"frontend_type",
"platform_id",
"menu_item_id",
name="uq_frontend_platform_menu_config",
),
# Unique constraint: one config per frontend+user+menu_item
UniqueConstraint(
"frontend_type",
"user_id",
"menu_item_id",
name="uq_frontend_user_menu_config",
),
# Check: exactly one scope must be set (platform_id XOR user_id)
CheckConstraint(
"(platform_id IS NOT NULL AND user_id IS NULL) OR "
"(platform_id IS NULL AND user_id IS NOT NULL)",
name="ck_admin_menu_config_scope",
),
# Check: user_id scope only allowed for admin frontend
CheckConstraint(
"(user_id IS NULL) OR (frontend_type = 'admin')",
name="ck_user_scope_admin_only",
),
# Performance indexes
Index(
"idx_admin_menu_config_frontend_platform",
"frontend_type",
"platform_id",
),
Index(
"idx_admin_menu_config_frontend_user",
"frontend_type",
"user_id",
),
Index(
"idx_admin_menu_config_platform_visible",
"platform_id",
"is_visible",
),
Index(
"idx_admin_menu_config_user_visible",
"user_id",
"is_visible",
),
)
# ========================================================================
# Properties
# ========================================================================
@property
def scope_type(self) -> str:
"""Get the scope type for this config."""
if self.platform_id:
return "platform"
return "user"
@property
def scope_id(self) -> int:
"""Get the scope ID (platform_id or user_id)."""
return self.platform_id or self.user_id
def __repr__(self) -> str:
scope = f"platform_id={self.platform_id}" if self.platform_id else f"user_id={self.user_id}"
return (
f"<AdminMenuConfig("
f"frontend_type='{self.frontend_type.value}', "
f"{scope}, "
f"menu_item_id='{self.menu_item_id}', "
f"is_visible={self.is_visible})>"
)

161
models/database/admin_platform.py
View File

@@ -1,161 +0,0 @@
# models/database/admin_platform.py
"""
AdminPlatform junction table for many-to-many relationship between Admin Users and Platforms.
This enables platform-scoped admin access:
- Super Admins: Have is_super_admin=True on User model, bypass this table
- Platform Admins: Assigned to specific platforms via this junction table
A platform admin CAN be assigned to multiple platforms (e.g., both OMS and Loyalty).
"""
from datetime import UTC, datetime
from sqlalchemy import (
Boolean,
Column,
DateTime,
ForeignKey,
Index,
Integer,
UniqueConstraint,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class AdminPlatform(Base, TimestampMixin):
"""
Junction table linking admin users to platforms they can manage.
Allows a platform admin to:
- Manage specific platforms only (not all)
- Be assigned to multiple platforms
- Have assignment tracked for audit purposes
Example:
- User "john@example.com" (admin) can manage OMS platform only
- User "jane@example.com" (admin) can manage both OMS and Loyalty platforms
"""
__tablename__ = "admin_platforms"
id = Column(Integer, primary_key=True, index=True)
# ========================================================================
# Foreign Keys
# ========================================================================
user_id = Column(
Integer,
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="Reference to the admin user",
)
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=False,
index=True,
comment="Reference to the platform",
)
# ========================================================================
# Assignment Status
# ========================================================================
is_active = Column(
Boolean,
default=True,
nullable=False,
comment="Whether the admin assignment is active",
)
# ========================================================================
# Audit Fields
# ========================================================================
assigned_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
nullable=False,
comment="When the admin was assigned to this platform",
)
assigned_by_user_id = Column(
Integer,
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
comment="Super admin who made this assignment",
)
# ========================================================================
# Relationships
# ========================================================================
user = relationship(
"User",
foreign_keys=[user_id],
back_populates="admin_platforms",
)
platform = relationship(
"Platform",
back_populates="admin_platforms",
)
assigned_by = relationship(
"User",
foreign_keys=[assigned_by_user_id],
)
# ========================================================================
# Constraints & Indexes
# ========================================================================
__table_args__ = (
# Each admin can only be assigned to a platform once
UniqueConstraint(
"user_id",
"platform_id",
name="uq_admin_platform",
),
# Performance indexes
Index(
"idx_admin_platform_active",
"user_id",
"platform_id",
"is_active",
),
Index(
"idx_admin_platform_user_active",
"user_id",
"is_active",
),
)
# ========================================================================
# Properties
# ========================================================================
@property
def platform_code(self) -> str | None:
"""Get the platform code for this assignment."""
return self.platform.code if self.platform else None
@property
def platform_name(self) -> str | None:
"""Get the platform name for this assignment."""
return self.platform.name if self.platform else None
def __repr__(self) -> str:
return (
f"<AdminPlatform("
f"user_id={self.user_id}, "
f"platform_id={self.platform_id}, "
f"is_active={self.is_active})>"
)

106
models/database/company.py
View File

@@ -1,106 +0,0 @@
# models/database/company.py
"""
Company model representing the business entity that owns one or more vendor brands.
A Company represents the legal/business entity with contact information,
while Vendors represent the individual brands/storefronts operated by that company.
"""
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class Company(Base, TimestampMixin):
"""
Represents a company (business entity) in the system.
A company owns one or more vendor brands. All business/contact information
is stored at the company level to avoid duplication.
"""
__tablename__ = "companies"
# ========================================================================
# Basic Information
# ========================================================================
id = Column(Integer, primary_key=True, index=True)
"""Unique identifier for the company."""
name = Column(String, nullable=False, index=True)
"""Company legal/business name."""
description = Column(Text)
"""Optional description of the company."""
# ========================================================================
# Ownership
# ========================================================================
owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
"""Foreign key to the user who owns this company."""
# ========================================================================
# Contact Information
# ========================================================================
contact_email = Column(String, nullable=False)
"""Primary business contact email."""
contact_phone = Column(String)
"""Business phone number."""
website = Column(String)
"""Company website URL."""
# ========================================================================
# Business Details
# ========================================================================
business_address = Column(Text)
"""Physical business address."""
tax_number = Column(String)
"""Tax/VAT registration number."""
# ========================================================================
# Status Flags
# ========================================================================
is_active = Column(Boolean, default=True, nullable=False)
"""Whether the company is active. Affects all associated vendors."""
is_verified = Column(Boolean, default=False, nullable=False)
"""Whether the company has been verified by platform admins."""
# ========================================================================
# Relationships
# ========================================================================
owner = relationship("User", back_populates="owned_companies")
"""The user who owns this company."""
vendors = relationship(
"Vendor",
back_populates="company",
cascade="all, delete-orphan",
order_by="Vendor.name",
)
"""All vendor brands operated by this company."""
def __repr__(self):
"""String representation of the Company object."""
return f"<Company(id={self.id}, name='{self.name}', vendors={len(self.vendors) if self.vendors else 0})>"
# ========================================================================
# Helper Properties
# ========================================================================
@property
def vendor_count(self) -> int:
"""Get the number of vendors belonging to this company."""
return len(self.vendors) if self.vendors else 0
@property
def active_vendor_count(self) -> int:
"""Get the number of active vendors belonging to this company."""
if not self.vendors:
return 0
return sum(1 for v in self.vendors if v.is_active)

306
models/database/email.py
View File

@@ -1,306 +0,0 @@
# models/database/email.py
"""
Email system database models.
Provides:
- EmailTemplate: Multi-language email templates stored in database
- EmailLog: Email sending history and tracking
Platform vs Vendor Templates:
- Platform templates (EmailTemplate) are the defaults
- Vendors can override templates via VendorEmailTemplate
- Platform-only templates (is_platform_only=True) cannot be overridden
"""
import enum
import json
from datetime import datetime
from sqlalchemy import (
Boolean,
Column,
DateTime,
Enum,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import Session, relationship
from app.core.database import Base
from .base import TimestampMixin
class EmailCategory(str, enum.Enum):
"""Email template categories."""
AUTH = "auth" # signup, password reset, verification
ORDERS = "orders" # order confirmations, shipping
BILLING = "billing" # invoices, payment failures
SYSTEM = "system" # team invites, notifications
MARKETING = "marketing" # newsletters, promotions
class EmailStatus(str, enum.Enum):
"""Email sending status."""
PENDING = "pending"
SENT = "sent"
FAILED = "failed"
BOUNCED = "bounced"
DELIVERED = "delivered"
OPENED = "opened"
CLICKED = "clicked"
class EmailTemplate(Base, TimestampMixin):
"""
Multi-language email templates.
Templates use Jinja2 syntax for variable interpolation.
Each template can have multiple language versions.
"""
__tablename__ = "email_templates"
id = Column(Integer, primary_key=True, index=True)
# Template identification
code = Column(String(100), nullable=False, index=True) # e.g., "signup_welcome"
language = Column(String(5), nullable=False, default="en") # e.g., "en", "fr", "de", "lb"
# Template metadata
name = Column(String(255), nullable=False) # Human-readable name
description = Column(Text, nullable=True) # Template purpose description
category = Column(
String(50), default=EmailCategory.SYSTEM.value, nullable=False, index=True
)
# Email content
subject = Column(String(500), nullable=False) # Subject line (supports variables)
body_html = Column(Text, nullable=False) # HTML body
body_text = Column(Text, nullable=True) # Plain text fallback
# Template variables (JSON list of expected variables)
# e.g., ["first_name", "company_name", "login_url"]
variables = Column(Text, nullable=True)
# Required variables (JSON list of variables that MUST be provided)
# Subset of variables that are mandatory for the template to render
required_variables = Column(Text, nullable=True)
# Status
is_active = Column(Boolean, default=True, nullable=False)
# Platform-only flag: if True, vendors cannot override this template
# Used for billing, subscription, and other platform-level emails
is_platform_only = Column(Boolean, default=False, nullable=False)
# Unique constraint: one template per code+language
__table_args__ = (
Index("ix_email_templates_code_language", "code", "language", unique=True),
{"sqlite_autoincrement": True},
)
def __repr__(self):
return f"<EmailTemplate(code='{self.code}', language='{self.language}')>"
@property
def variables_list(self) -> list[str]:
"""Parse variables JSON to list."""
if not self.variables:
return []
try:
return json.loads(self.variables)
except (json.JSONDecodeError, TypeError):
return []
@property
def required_variables_list(self) -> list[str]:
"""Parse required_variables JSON to list."""
if not self.required_variables:
return []
try:
return json.loads(self.required_variables)
except (json.JSONDecodeError, TypeError):
return []
@classmethod
def get_by_code_and_language(
cls,
db: Session,
code: str,
language: str,
fallback_to_english: bool = True,
) -> "EmailTemplate | None":
"""
Get a platform template by code and language.
Args:
db: Database session
code: Template code (e.g., "password_reset")
language: Language code (en, fr, de, lb)
fallback_to_english: If True, fall back to English if language not found
Returns:
EmailTemplate if found, None otherwise
"""
template = (
db.query(cls)
.filter(
cls.code == code,
cls.language == language,
cls.is_active == True, # noqa: E712
)
.first()
)
# Fallback to English if requested language not found
if not template and fallback_to_english and language != "en":
template = (
db.query(cls)
.filter(
cls.code == code,
cls.language == "en",
cls.is_active == True, # noqa: E712
)
.first()
)
return template
@classmethod
def get_all_templates(
cls,
db: Session,
category: str | None = None,
include_inactive: bool = False,
) -> list["EmailTemplate"]:
"""
Get all platform templates, optionally filtered by category.
Args:
db: Database session
category: Optional category filter
include_inactive: Include inactive templates
Returns:
List of EmailTemplate objects
"""
query = db.query(cls)
if category:
query = query.filter(cls.category == category)
if not include_inactive:
query = query.filter(cls.is_active == True) # noqa: E712
return query.order_by(cls.code, cls.language).all()
@classmethod
def get_overridable_templates(cls, db: Session) -> list["EmailTemplate"]:
"""
Get all templates that vendors can override.
Returns:
List of EmailTemplate objects where is_platform_only=False
"""
return (
db.query(cls)
.filter(
cls.is_platform_only == False, # noqa: E712
cls.is_active == True, # noqa: E712
)
.order_by(cls.code, cls.language)
.all()
)
class EmailLog(Base, TimestampMixin):
"""
Email sending history and tracking.
Logs all sent emails for debugging, analytics, and compliance.
"""
__tablename__ = "email_logs"
id = Column(Integer, primary_key=True, index=True)
# Template reference
template_code = Column(String(100), nullable=True, index=True)
template_id = Column(Integer, ForeignKey("email_templates.id"), nullable=True)
# Recipient info
recipient_email = Column(String(255), nullable=False, index=True)
recipient_name = Column(String(255), nullable=True)
# Email content (snapshot at send time)
subject = Column(String(500), nullable=False)
body_html = Column(Text, nullable=True)
body_text = Column(Text, nullable=True)
# Sending info
from_email = Column(String(255), nullable=False)
from_name = Column(String(255), nullable=True)
reply_to = Column(String(255), nullable=True)
# Status tracking
status = Column(
String(20), default=EmailStatus.PENDING.value, nullable=False, index=True
)
sent_at = Column(DateTime, nullable=True)
delivered_at = Column(DateTime, nullable=True)
opened_at = Column(DateTime, nullable=True)
clicked_at = Column(DateTime, nullable=True)
# Error handling
error_message = Column(Text, nullable=True)
retry_count = Column(Integer, default=0, nullable=False)
# Provider info
provider = Column(String(50), nullable=True) # smtp, sendgrid, mailgun, ses
provider_message_id = Column(String(255), nullable=True, index=True)
# Context linking (optional - link to related entities)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
related_type = Column(String(50), nullable=True) # e.g., "order", "subscription"
related_id = Column(Integer, nullable=True)
# Extra data (JSON for additional context)
extra_data = Column(Text, nullable=True)
# Relationships
template = relationship("EmailTemplate", foreign_keys=[template_id])
vendor = relationship("Vendor", foreign_keys=[vendor_id])
user = relationship("User", foreign_keys=[user_id])
def __repr__(self):
return f"<EmailLog(id={self.id}, recipient='{self.recipient_email}', status='{self.status}')>"
def mark_sent(self, provider_message_id: str | None = None):
"""Mark email as sent."""
self.status = EmailStatus.SENT.value
self.sent_at = datetime.utcnow()
if provider_message_id:
self.provider_message_id = provider_message_id
def mark_failed(self, error_message: str):
"""Mark email as failed."""
self.status = EmailStatus.FAILED.value
self.error_message = error_message
self.retry_count += 1
def mark_delivered(self):
"""Mark email as delivered."""
self.status = EmailStatus.DELIVERED.value
self.delivered_at = datetime.utcnow()
def mark_opened(self):
"""Mark email as opened."""
self.status = EmailStatus.OPENED.value
self.opened_at = datetime.utcnow()

127
models/database/media.py
View File

@@ -1,127 +0,0 @@
# models/database/media.py
"""
CORE media file model for vendor media library.
This is a CORE framework model used across multiple modules.
MediaFile provides vendor-uploaded media files (images, documents, videos).
For product-media associations, use:
from app.modules.catalog.models import ProductMedia
Files are stored in vendor-specific directories:
uploads/vendors/{vendor_id}/{folder}/{filename}
"""
from sqlalchemy import (
Boolean,
Column,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.dialects.sqlite import JSON
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class MediaFile(Base, TimestampMixin):
"""Vendor media file record.
Stores metadata about uploaded files. Actual files are stored
in the filesystem at uploads/vendors/{vendor_id}/{folder}/
"""
__tablename__ = "media_files"
id = Column(Integer, primary_key=True, index=True)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
# File identification
filename = Column(String(255), nullable=False) # Stored filename (UUID-based)
original_filename = Column(String(255)) # Original uploaded filename
file_path = Column(String(500), nullable=False) # Relative path from uploads/
# File properties
media_type = Column(String(20), nullable=False) # image, video, document
mime_type = Column(String(100))
file_size = Column(Integer) # bytes
# Image/video dimensions
width = Column(Integer)
height = Column(Integer)
# Thumbnail (for images/videos)
thumbnail_path = Column(String(500))
# Metadata
alt_text = Column(String(500))
description = Column(Text)
folder = Column(String(100), default="general") # products, general, etc.
tags = Column(JSON) # List of tags for categorization
extra_metadata = Column(JSON) # Additional metadata (EXIF, etc.)
# Status
is_optimized = Column(Boolean, default=False)
optimized_size = Column(Integer) # Size after optimization
# Usage tracking
usage_count = Column(Integer, default=0) # How many times used
# Relationships
vendor = relationship("Vendor", back_populates="media_files")
# ProductMedia relationship uses string reference to avoid circular import
product_associations = relationship(
"ProductMedia",
back_populates="media",
cascade="all, delete-orphan",
)
__table_args__ = (
Index("idx_media_vendor_id", "vendor_id"),
Index("idx_media_vendor_folder", "vendor_id", "folder"),
Index("idx_media_vendor_type", "vendor_id", "media_type"),
Index("idx_media_filename", "filename"),
)
def __repr__(self):
return (
f"<MediaFile(id={self.id}, vendor_id={self.vendor_id}, "
f"filename='{self.filename}', type='{self.media_type}')>"
)
@property
def file_url(self) -> str:
"""Get the public URL for this file."""
return f"/uploads/{self.file_path}"
@property
def thumbnail_url(self) -> str | None:
"""Get the thumbnail URL if available."""
if self.thumbnail_path:
return f"/uploads/{self.thumbnail_path}"
return None
@property
def is_image(self) -> bool:
"""Check if this is an image file."""
return self.media_type == "image"
@property
def is_video(self) -> bool:
"""Check if this is a video file."""
return self.media_type == "video"
@property
def is_document(self) -> bool:
"""Check if this is a document file."""
return self.media_type == "document"
# Re-export ProductMedia from its canonical location for backwards compatibility
from app.modules.catalog.models import ProductMedia # noqa: E402, F401
__all__ = ["MediaFile", "ProductMedia"]

239
models/database/platform.py
View File

@@ -1,239 +0,0 @@
# 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",
)
# Admin assignments for this platform
admin_platforms = relationship(
"AdminPlatform",
back_populates="platform",
cascade="all, delete-orphan",
)
# Menu visibility configuration for platform admins
menu_configs = relationship(
"AdminMenuConfig",
back_populates="platform",
cascade="all, delete-orphan",
)
# Module enablement configuration
modules = relationship(
"PlatformModule",
back_populates="platform",
cascade="all, delete-orphan",
)
# ========================================================================
# 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}')>"

162
models/database/platform_module.py
View File

@@ -1,162 +0,0 @@
# models/database/platform_module.py
"""
PlatformModule model for tracking module enablement per platform.
This junction table provides:
- Auditability: Track when modules were enabled/disabled and by whom
- Configuration: Per-module settings specific to each platform
- State tracking: Explicit enabled/disabled states with timestamps
Replaces the simpler Platform.settings["enabled_modules"] JSON approach
for better auditability and query capabilities.
"""
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 PlatformModule(Base, TimestampMixin):
"""
Junction table tracking module enablement per platform.
This provides a normalized, auditable way to track which modules
are enabled for each platform, with configuration options.
Example:
PlatformModule(
platform_id=1,
module_code="billing",
is_enabled=True,
enabled_at=datetime.now(),
enabled_by_user_id=42,
config={"stripe_mode": "live", "default_trial_days": 14}
)
"""
__tablename__ = "platform_modules"
id = Column(Integer, primary_key=True, index=True)
# ========================================================================
# Identity
# ========================================================================
platform_id = Column(
Integer,
ForeignKey("platforms.id", ondelete="CASCADE"),
nullable=False,
comment="Platform this module configuration belongs to",
)
module_code = Column(
String(50),
nullable=False,
comment="Module code (e.g., 'billing', 'inventory', 'orders')",
)
# ========================================================================
# State
# ========================================================================
is_enabled = Column(
Boolean,
nullable=False,
default=True,
comment="Whether this module is currently enabled for the platform",
)
# ========================================================================
# Audit Trail - Enable
# ========================================================================
enabled_at = Column(
DateTime(timezone=True),
nullable=True,
comment="When the module was last enabled",
)
enabled_by_user_id = Column(
Integer,
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
comment="User who enabled the module",
)
# ========================================================================
# Audit Trail - Disable
# ========================================================================
disabled_at = Column(
DateTime(timezone=True),
nullable=True,
comment="When the module was last disabled",
)
disabled_by_user_id = Column(
Integer,
ForeignKey("users.id", ondelete="SET NULL"),
nullable=True,
comment="User who disabled the module",
)
# ========================================================================
# Configuration
# ========================================================================
config = Column(
JSON,
nullable=False,
default=dict,
comment="Module-specific configuration for this platform",
)
# ========================================================================
# Relationships
# ========================================================================
platform = relationship(
"Platform",
back_populates="modules",
)
enabled_by = relationship(
"User",
foreign_keys=[enabled_by_user_id],
)
disabled_by = relationship(
"User",
foreign_keys=[disabled_by_user_id],
)
# ========================================================================
# Constraints & Indexes
# ========================================================================
__table_args__ = (
# Each platform can only have one configuration per module
UniqueConstraint("platform_id", "module_code", name="uq_platform_module"),
# Index for querying by platform
Index("idx_platform_module_platform_id", "platform_id"),
# Index for querying by module code
Index("idx_platform_module_code", "module_code"),
# Index for querying enabled modules
Index("idx_platform_module_enabled", "platform_id", "is_enabled"),
)
def __repr__(self) -> str:
status = "enabled" if self.is_enabled else "disabled"
return f"<PlatformModule(platform_id={self.platform_id}, module='{self.module_code}', {status})>"

199
models/database/user.py
View File

@@ -1,199 +0,0 @@
# models/database/user.py - IMPROVED VERSION
"""
User model with authentication support.
ROLE CLARIFICATION:
- User.role should ONLY contain platform-level roles:
* "admin" - Platform administrator (full system access)
* "vendor" - Any user who owns or is part of a vendor team
- Vendor-specific roles (manager, staff, etc.) are stored in VendorUser.role
- Customers are NOT in the User table - they use the Customer model
"""
import enum
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class UserRole(str, enum.Enum):
"""Platform-level user roles."""
ADMIN = "admin" # Platform administrator
VENDOR = "vendor" # Vendor owner or team member
class User(Base, TimestampMixin):
"""Represents a platform user (admins and vendors only)."""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
username = Column(String, unique=True, index=True, nullable=False)
first_name = Column(String)
last_name = Column(String)
hashed_password = Column(String, nullable=False)
# Platform-level role only (admin or vendor)
role = Column(String, nullable=False, default=UserRole.VENDOR.value)
is_active = Column(Boolean, default=True, nullable=False)
is_email_verified = Column(Boolean, default=False, nullable=False)
last_login = Column(DateTime, nullable=True)
# Super admin flag (only meaningful when role='admin')
# Super admins have access to ALL platforms and global settings
# Platform admins (is_super_admin=False) are assigned to specific platforms
is_super_admin = Column(Boolean, default=False, nullable=False)
# Language preference (NULL = use context default: vendor dashboard_language or system default)
# Supported: en, fr, de, lb
preferred_language = Column(String(5), nullable=True)
# Relationships
marketplace_import_jobs = relationship(
"MarketplaceImportJob", back_populates="user"
)
owned_companies = relationship("Company", back_populates="owner")
vendor_memberships = relationship(
"VendorUser", foreign_keys="[VendorUser.user_id]", back_populates="user"
)
# Admin-platform assignments (for platform admins only)
# Super admins don't need assignments - they have access to all platforms
admin_platforms = relationship(
"AdminPlatform",
foreign_keys="AdminPlatform.user_id",
back_populates="user",
cascade="all, delete-orphan",
)
# Menu visibility configuration (for super admins only)
# Platform admins get menu config from their platform, not user-level
menu_configs = relationship(
"AdminMenuConfig",
foreign_keys="AdminMenuConfig.user_id",
back_populates="user",
cascade="all, delete-orphan",
)
def __repr__(self):
"""String representation of the User object."""
return f"<User(id={self.id}, username='{self.username}', email='{self.email}', role='{self.role}')>"
@property
def full_name(self):
"""Returns the full name of the user, combining first and last names if available."""
if self.first_name and self.last_name:
return f"{self.first_name} {self.last_name}"
return self.username
@property
def is_admin(self) -> bool:
"""Check if user is a platform admin."""
return self.role == UserRole.ADMIN.value
@property
def is_vendor(self) -> bool:
"""Check if user is a vendor (owner or team member)."""
return self.role == UserRole.VENDOR.value
def is_owner_of(self, vendor_id: int) -> bool:
"""
Check if user is the owner of a specific vendor.
Ownership is determined via company ownership:
User owns Company → Company has Vendor → User owns Vendor
"""
for company in self.owned_companies:
if any(v.id == vendor_id for v in company.vendors):
return True
return False
def is_member_of(self, vendor_id: int) -> bool:
"""Check if user is a member of a specific vendor (owner or team)."""
# Check if owner (via company)
if self.is_owner_of(vendor_id):
return True
# Check if team member
return any(
vm.vendor_id == vendor_id and vm.is_active for vm in self.vendor_memberships
)
def get_vendor_role(self, vendor_id: int) -> str:
"""Get user's role within a specific vendor."""
# Check if owner (via company)
if self.is_owner_of(vendor_id):
return "owner"
# Check team membership
for vm in self.vendor_memberships:
if vm.vendor_id == vendor_id and vm.is_active:
return vm.role.name if vm.role else "member"
return None
def has_vendor_permission(self, vendor_id: int, permission: str) -> bool:
"""Check if user has a specific permission in a vendor."""
# Owners have all permissions
if self.is_owner_of(vendor_id):
return True
# Check team member permissions
for vm in self.vendor_memberships:
if vm.vendor_id == vendor_id and vm.is_active:
if vm.role and permission in vm.role.permissions:
return True
return False
# =========================================================================
# Admin Platform Access Methods
# =========================================================================
@property
def is_super_admin_user(self) -> bool:
"""Check if user is a super admin (can access all platforms)."""
return self.role == UserRole.ADMIN.value and self.is_super_admin
@property
def is_platform_admin(self) -> bool:
"""Check if user is a platform admin (access to assigned platforms only)."""
return self.role == UserRole.ADMIN.value and not self.is_super_admin
def can_access_platform(self, platform_id: int) -> bool:
"""
Check if admin can access a specific platform.
- Super admins can access all platforms
- Platform admins can only access assigned platforms
- Non-admins return False
"""
if not self.is_admin:
return False
if self.is_super_admin:
return True
return any(
ap.platform_id == platform_id and ap.is_active
for ap in self.admin_platforms
)
def get_accessible_platform_ids(self) -> list[int] | None:
"""
Get list of platform IDs this admin can access.
Returns:
- None for super admins (means ALL platforms)
- List of platform IDs for platform admins
- Empty list for non-admins
"""
if not self.is_admin:
return []
if self.is_super_admin:
return None # None means ALL platforms
return [ap.platform_id for ap in self.admin_platforms if ap.is_active]

568
models/database/vendor.py
View File

@@ -1,568 +0,0 @@
# models/database/vendor.py
"""
Vendor model representing entities that sell products or services.
This module defines the Vendor model along with its relationships to
other models such as User (owner), Product, Customer, Order, and MarketplaceImportJob.
"""
import enum
from sqlalchemy import (
JSON,
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.config import settings
# Import Base from the central database module instead of creating a new one
from app.core.database import Base
from models.database.base import TimestampMixin
class Vendor(Base, TimestampMixin):
"""Represents a vendor in the system."""
__tablename__ = "vendors" # Name of the table in the database
id = Column(
Integer, primary_key=True, index=True
) # Primary key and indexed column for vendor ID
# Company relationship
company_id = Column(
Integer, ForeignKey("companies.id"), nullable=False, index=True
) # Foreign key to the parent company
vendor_code = Column(
String, unique=True, index=True, nullable=False
) # Unique, indexed, non-nullable vendor code column
subdomain = Column(
String(100), unique=True, nullable=False, index=True
) # Unique, non-nullable subdomain column with indexing
name = Column(
String, nullable=False
) # Non-nullable name column for the vendor (brand name)
description = Column(Text) # Optional text description column for the vendor
# Letzshop URLs - multi-language support (brand-specific marketplace feeds)
letzshop_csv_url_fr = Column(String) # URL for French CSV in Letzshop
letzshop_csv_url_en = Column(String) # URL for English CSV in Letzshop
letzshop_csv_url_de = Column(String) # URL for German CSV in Letzshop
# Letzshop Vendor Identity (for linking to Letzshop marketplace profile)
letzshop_vendor_id = Column(
String(100), unique=True, nullable=True, index=True
) # Letzshop's vendor identifier
letzshop_vendor_slug = Column(
String(200), nullable=True, index=True
) # Letzshop shop URL slug (e.g., "my-shop" from letzshop.lu/vendors/my-shop)
# ========================================================================
# Letzshop Feed Settings (atalanda namespace)
# ========================================================================
# These are default values applied to all products in the Letzshop feed
# See https://letzshop.lu/en/dev#google_csv for documentation
# Default VAT rate for new products: 0 (exempt), 3 (super-reduced), 8 (reduced), 14 (intermediate), 17 (standard)
letzshop_default_tax_rate = Column(Integer, default=17, nullable=False)
# Product sort priority on Letzshop (0.0-10.0, higher = displayed first)
# Note: Having all products rated above 7 is not permitted by Letzshop
letzshop_boost_sort = Column(String(10), default="5.0") # Stored as string for precision
# Delivery method: 'nationwide', 'package_delivery', 'self_collect' (comma-separated for multiple)
# 'nationwide' automatically includes package_delivery and self_collect
letzshop_delivery_method = Column(String(100), default="package_delivery")
# Pre-order days: number of days before item ships (default 1 day)
letzshop_preorder_days = Column(Integer, default=1)
# Status (vendor-specific, can differ from company status)
is_active = Column(
Boolean, default=True
) # Boolean to indicate if the vendor brand is active
is_verified = Column(
Boolean, default=False
) # Boolean to indicate if the vendor brand is verified
# ========================================================================
# Contact Information (nullable = inherit from company)
# ========================================================================
# These fields allow vendor-specific branding/identity.
# If null, the value is inherited from the parent company.
contact_email = Column(String(255), nullable=True) # Override company contact email
contact_phone = Column(String(50), nullable=True) # Override company contact phone
website = Column(String(255), nullable=True) # Override company website
business_address = Column(Text, nullable=True) # Override company business address
tax_number = Column(String(100), nullable=True) # Override company tax number
# ========================================================================
# Language Settings
# ========================================================================
# Supported languages: en, fr, de, lb (Luxembourgish)
default_language = Column(
String(5), nullable=False, default="fr"
) # Default language for vendor content (products, emails, etc.)
dashboard_language = Column(
String(5), nullable=False, default="fr"
) # Language for vendor team dashboard UI
storefront_language = Column(
String(5), nullable=False, default="fr"
) # Default language for customer-facing storefront
storefront_languages = Column(
JSON, nullable=False, default=["fr", "de", "en"]
) # Array of enabled languages for storefront language selector
# Currency/number formatting locale (e.g., 'fr-LU' = "29,99 €", 'en-GB' = "€29.99")
# NULL means inherit from platform default (AdminSetting 'default_storefront_locale')
storefront_locale = Column(String(10), nullable=True)
# ========================================================================
# Relationships
# ========================================================================
company = relationship(
"Company", back_populates="vendors"
) # Relationship with Company model for the parent company
vendor_users = relationship(
"VendorUser", back_populates="vendor"
) # Relationship with VendorUser model for users in this vendor
products = relationship(
"Product", back_populates="vendor"
) # Relationship with Product model for products of this vendor
customers = relationship(
"Customer", back_populates="vendor"
) # Relationship with Customer model for customers of this vendor
orders = relationship(
"Order", back_populates="vendor"
) # Relationship with Order model for orders placed by this vendor
marketplace_import_jobs = relationship(
"MarketplaceImportJob", back_populates="vendor"
) # Relationship with MarketplaceImportJob model for import jobs related to this vendor
# Letzshop integration credentials (one-to-one)
letzshop_credentials = relationship(
"VendorLetzshopCredentials",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
)
# Invoice settings (one-to-one)
invoice_settings = relationship(
"VendorInvoiceSettings",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
)
# Invoices (one-to-many)
invoices = relationship(
"Invoice",
back_populates="vendor",
cascade="all, delete-orphan",
)
# Email template overrides (one-to-many)
email_templates = relationship(
"VendorEmailTemplate",
back_populates="vendor",
cascade="all, delete-orphan",
)
# Email settings (one-to-one) - vendor SMTP/provider configuration
email_settings = relationship(
"VendorEmailSettings",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
)
# Subscription (one-to-one)
subscription = relationship(
"VendorSubscription",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
)
# Add-ons purchased by vendor (one-to-many)
addons = relationship(
"VendorAddOn",
back_populates="vendor",
cascade="all, delete-orphan",
)
# Billing/invoice history (one-to-many)
billing_history = relationship(
"BillingHistory",
back_populates="vendor",
cascade="all, delete-orphan",
order_by="BillingHistory.invoice_date.desc()",
)
domains = relationship(
"VendorDomain",
back_populates="vendor",
cascade="all, delete-orphan",
order_by="VendorDomain.is_primary.desc()",
) # Relationship with VendorDomain model for custom domains of the vendor
# Single theme relationship (ONE vendor = ONE theme)
# A vendor has ONE active theme stored in the vendor_themes table.
# Theme presets available: default, modern, classic, minimal, vibrant
vendor_theme = relationship(
"VendorTheme",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
) # Relationship with VendorTheme model for the active theme of the vendor
# Content pages relationship (vendor can override platform default pages)
content_pages = relationship(
"ContentPage", back_populates="vendor", cascade="all, delete-orphan"
) # Relationship with ContentPage model for vendor-specific content pages
# Onboarding progress (one-to-one)
onboarding = relationship(
"VendorOnboarding",
back_populates="vendor",
uselist=False,
cascade="all, delete-orphan",
)
# Media library (one-to-many)
media_files = relationship(
"MediaFile",
back_populates="vendor",
cascade="all, delete-orphan",
)
# Platform memberships (many-to-many via junction table)
vendor_platforms = relationship(
"VendorPlatform",
back_populates="vendor",
cascade="all, delete-orphan",
)
# Loyalty program (one-to-one)
loyalty_program = relationship(
"LoyaltyProgram",
back_populates="vendor",
uselist=False,
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}')>"
# ========================================================================
# Theme Helper Methods to get active theme and other related information
# ========================================================================
def get_effective_theme(self) -> dict:
"""
Get active theme for this vendor.
Returns theme from vendor_themes table, or default theme if not set.
Returns:
dict: Theme configuration with colors, fonts, layout, etc.
"""
# Check vendor_themes table
if self.vendor_theme and self.vendor_theme.is_active:
return self.vendor_theme.to_dict()
# Return default theme
return self._get_default_theme()
def _get_default_theme(self) -> dict:
"""Return the default theme configuration."""
return {
"theme_name": "default",
"colors": {
"primary": "#6366f1",
"secondary": "#8b5cf6",
"accent": "#ec4899",
"background": "#ffffff",
"text": "#1f2937",
"border": "#e5e7eb",
},
"fonts": {"heading": "Inter, sans-serif", "body": "Inter, sans-serif"},
"branding": {
"logo": None,
"logo_dark": None,
"favicon": None,
"banner": None,
},
"layout": {"style": "grid", "header": "fixed", "product_card": "modern"},
"social_links": {},
"custom_css": None,
"css_variables": {
"--color-primary": "#6366f1",
"--color-secondary": "#8b5cf6",
"--color-accent": "#ec4899",
"--color-background": "#ffffff",
"--color-text": "#1f2937",
"--color-border": "#e5e7eb",
"--font-heading": "Inter, sans-serif",
"--font-body": "Inter, sans-serif",
},
}
def get_primary_color(self) -> str:
"""Get primary color from active theme."""
theme = self.get_effective_theme()
return theme.get("colors", {}).get(
"primary", "#6366f1"
) # Default to default theme if not found
def get_logo_url(self) -> str:
"""Get logo URL from active theme."""
theme = self.get_effective_theme()
return theme.get("branding", {}).get(
"logo"
) # Return None or the logo URL if found
# ========================================================================
# Domain Helper Methods
# ========================================================================
@property
def primary_domain(self):
"""Get the primary custom domain for this vendor."""
for domain in self.domains:
if domain.is_primary and domain.is_active:
return domain.domain # Return the domain if it's primary and active
return None
@property
def all_domains(self):
"""Get all active domains (subdomain + custom domains)."""
domains = [
f"{self.subdomain}.{settings.platform_domain}"
] # Start with the main subdomain
for domain in self.domains:
if domain.is_active:
domains.append(domain.domain) # Add other active custom domains
return domains
# ========================================================================
# Contact Resolution Helper Properties
# ========================================================================
# These properties return the effective value (vendor override or company fallback)
@property
def effective_contact_email(self) -> str | None:
"""Get contact email (vendor override or company fallback)."""
if self.contact_email is not None:
return self.contact_email
return self.company.contact_email if self.company else None
@property
def effective_contact_phone(self) -> str | None:
"""Get contact phone (vendor override or company fallback)."""
if self.contact_phone is not None:
return self.contact_phone
return self.company.contact_phone if self.company else None
@property
def effective_website(self) -> str | None:
"""Get website (vendor override or company fallback)."""
if self.website is not None:
return self.website
return self.company.website if self.company else None
@property
def effective_business_address(self) -> str | None:
"""Get business address (vendor override or company fallback)."""
if self.business_address is not None:
return self.business_address
return self.company.business_address if self.company else None
@property
def effective_tax_number(self) -> str | None:
"""Get tax number (vendor override or company fallback)."""
if self.tax_number is not None:
return self.tax_number
return self.company.tax_number if self.company else None
def get_contact_info_with_inheritance(self) -> dict:
"""
Get all contact info with inheritance flags.
Returns dict with resolved values and flags indicating if inherited from company.
"""
company = self.company
return {
"contact_email": self.effective_contact_email,
"contact_email_inherited": self.contact_email is None
and company is not None,
"contact_phone": self.effective_contact_phone,
"contact_phone_inherited": self.contact_phone is None
and company is not None,
"website": self.effective_website,
"website_inherited": self.website is None and company is not None,
"business_address": self.effective_business_address,
"business_address_inherited": self.business_address is None
and company is not None,
"tax_number": self.effective_tax_number,
"tax_number_inherited": self.tax_number is None and company is not None,
}
class VendorUserType(str, enum.Enum):
"""Types of vendor users."""
OWNER = "owner" # Vendor owner (full access to vendor area)
TEAM_MEMBER = "member" # Team member (role-based access to vendor area)
class VendorUser(Base, TimestampMixin):
"""
Represents a user's membership in a vendor.
- Owner: Created automatically when vendor is created
- Team Member: Invited by owner via email
"""
__tablename__ = "vendor_users"
id = Column(Integer, primary_key=True, index=True)
"""Unique identifier for each VendorUser entry."""
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
"""Foreign key linking to the associated Vendor."""
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
"""Foreign key linking to the associated User."""
# Distinguish between owner and team member
user_type = Column(String, nullable=False, default=VendorUserType.TEAM_MEMBER.value)
# Role for team members (NULL for owners - they have all permissions)
role_id = Column(Integer, ForeignKey("roles.id"), nullable=True)
"""Foreign key linking to the associated Role."""
invited_by = Column(Integer, ForeignKey("users.id"))
"""Foreign key linking to the user who invited this VendorUser."""
invitation_token = Column(String, nullable=True, index=True) # For email activation
invitation_sent_at = Column(DateTime, nullable=True)
invitation_accepted_at = Column(DateTime, nullable=True)
is_active = Column(
Boolean, default=False, nullable=False
) # False until invitation accepted
"""Indicates whether the VendorUser role is active."""
# Relationships
vendor = relationship("Vendor", back_populates="vendor_users")
"""Relationship to the Vendor model, representing the associated vendor."""
user = relationship(
"User", foreign_keys=[user_id], back_populates="vendor_memberships"
)
"""Relationship to the User model, representing the user who holds this role within the vendor."""
inviter = relationship("User", foreign_keys=[invited_by])
"""Optional relationship to the User model, representing the user who invited this VendorUser."""
role = relationship("Role", back_populates="vendor_users")
"""Relationship to the Role model, representing the role held by the vendor user."""
def __repr__(self) -> str:
"""Return a string representation of the VendorUser instance.
Returns:
str: A string that includes the vendor_id, the user_id and the user_type of the VendorUser instance.
"""
return f"<VendorUser(vendor_id={self.vendor_id}, user_id={self.user_id}, type={self.user_type})>"
@property
def is_owner(self) -> bool:
"""Check if this is an owner membership."""
return self.user_type == VendorUserType.OWNER.value
@property
def is_team_member(self) -> bool:
"""Check if this is a team member (not owner)."""
return self.user_type == VendorUserType.TEAM_MEMBER.value
@property
def is_invitation_pending(self) -> bool:
"""Check if invitation is still pending."""
return self.invitation_token is not None and self.invitation_accepted_at is None
def has_permission(self, permission: str) -> bool:
"""
Check if user has a specific permission.
Owners always have all permissions.
Team members check their role's permissions.
"""
# Owners have all permissions
if self.is_owner:
return True
# Inactive users have no permissions
if not self.is_active:
return False
# Check role permissions
if self.role and self.role.permissions:
return permission in self.role.permissions
return False
def get_all_permissions(self) -> list:
"""Get all permissions this user has."""
if self.is_owner:
# Return all possible permissions
from app.core.permissions import VendorPermissions
return list(VendorPermissions.__members__.values())
if self.role and self.role.permissions:
return self.role.permissions
return []
class Role(Base, TimestampMixin):
"""Represents a role within a vendor's system."""
__tablename__ = "roles" # Name of the table in the database
id = Column(Integer, primary_key=True, index=True)
"""Unique identifier for each Role entry."""
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
"""Foreign key linking to the associated Vendor."""
name = Column(String(100), nullable=False)
"""Name of the role, with a maximum length of 100 characters."""
permissions = Column(JSON, default=list)
"""Permissions assigned to this role, stored as a JSON array."""
vendor = relationship("Vendor")
"""Relationship to the Vendor model, representing the associated vendor."""
vendor_users = relationship("VendorUser", back_populates="role")
"""Back-relationship to the VendorUser model, representing users with this role."""
def __repr__(self) -> str:
"""Return a string representation of the Role instance.
Returns:
str: A string that includes the id and name of the Role instance.
"""
return f"<Role(id={self.id}, name='{self.name}', vendor_id={self.vendor_id})>"

96
models/database/vendor_domain.py
View File

@@ -1,96 +0,0 @@
# models/database/vendor_domain.py
"""
Vendor Domain Model - Maps custom domains to vendors
"""
from sqlalchemy import (
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 VendorDomain(Base, TimestampMixin):
"""
Maps custom domains to vendors for multi-domain routing.
Examples:
- customdomain1.com → Vendor 1
- shop.mybusiness.com → Vendor 2
- www.customdomain1.com → Vendor 1 (www is stripped)
"""
__tablename__ = "vendor_domains"
id = Column(Integer, primary_key=True, index=True)
vendor_id = Column(
Integer, ForeignKey("vendors.id", ondelete="CASCADE"), nullable=False
)
# Domain configuration
domain = Column(String(255), nullable=False, unique=True, index=True)
is_primary = Column(Boolean, default=False, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
# SSL/TLS status (for monitoring)
ssl_status = Column(
String(50), default="pending"
) # pending, active, expired, error
ssl_verified_at = Column(DateTime(timezone=True), nullable=True)
# DNS verification (to confirm domain ownership)
verification_token = Column(String(100), unique=True, nullable=True)
is_verified = Column(Boolean, default=False, nullable=False)
verified_at = Column(DateTime(timezone=True), nullable=True)
# Relationships
vendor = relationship("Vendor", back_populates="domains")
# Constraints
__table_args__ = (
UniqueConstraint("vendor_id", "domain", name="uq_vendor_domain"),
Index("idx_domain_active", "domain", "is_active"),
Index("idx_vendor_primary", "vendor_id", "is_primary"),
)
def __repr__(self):
return f"<VendorDomain(domain='{self.domain}', vendor_id={self.vendor_id})>"
@property
def full_url(self):
"""Return full URL with https"""
return f"https://{self.domain}"
@classmethod
def normalize_domain(cls, domain: str) -> str:
"""
Normalize domain for consistent storage.
Examples:
- https://example.com → example.com
- www.example.com → example.com
- EXAMPLE.COM → example.com
"""
# Remove protocol
domain = domain.replace("https://", "").replace("http://", "") # noqa: SEC-034
# Remove trailing slash
domain = domain.rstrip("/")
# Remove www prefix (optional - depends on your preference)
# if domain.startswith("www."):
# domain = domain[4:]
# Convert to lowercase
domain = domain.lower()
return domain

255
models/database/vendor_email_settings.py
View File

@@ -1,255 +0,0 @@
# models/database/vendor_email_settings.py
"""
Vendor Email Settings model for vendor-specific email configuration.
This model stores vendor SMTP/email provider settings, enabling vendors to:
- Send emails from their own domain/email address
- Use their own SMTP server or email provider (tier-gated)
- Customize sender name, reply-to address, and signature
Architecture:
- Vendors MUST configure email settings to send transactional emails
- Platform emails (billing, subscription) still use platform settings
- Advanced providers (SendGrid, Mailgun, SES) are tier-gated (Business+)
- "Powered by Wizamart" footer is added for Essential/Professional tiers
"""
import enum
from datetime import UTC, datetime
from sqlalchemy import (
Boolean,
Column,
DateTime,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class EmailProvider(str, enum.Enum):
"""Supported email providers."""
SMTP = "smtp" # Standard SMTP (all tiers)
SENDGRID = "sendgrid" # SendGrid API (Business+ tier)
MAILGUN = "mailgun" # Mailgun API (Business+ tier)
SES = "ses" # Amazon SES (Business+ tier)
# Providers that require Business+ tier
PREMIUM_EMAIL_PROVIDERS = {
EmailProvider.SENDGRID,
EmailProvider.MAILGUN,
EmailProvider.SES,
}
class VendorEmailSettings(Base, TimestampMixin):
"""
Vendor email configuration for sending transactional emails.
This is a one-to-one relationship with Vendor.
Vendors must configure this to send emails to their customers.
"""
__tablename__ = "vendor_email_settings"
id = Column(Integer, primary_key=True, index=True)
vendor_id = Column(
Integer,
ForeignKey("vendors.id", ondelete="CASCADE"),
unique=True,
nullable=False,
index=True,
)
# =========================================================================
# Sender Identity (Required)
# =========================================================================
from_email = Column(String(255), nullable=False) # e.g., orders@vendorshop.lu
from_name = Column(String(100), nullable=False) # e.g., "VendorShop"
reply_to_email = Column(String(255), nullable=True) # Optional reply-to address
# =========================================================================
# Email Signature/Footer (Optional)
# =========================================================================
signature_text = Column(Text, nullable=True) # Plain text signature
signature_html = Column(Text, nullable=True) # HTML signature (footer)
# =========================================================================
# Provider Configuration
# =========================================================================
provider = Column(
String(20),
default=EmailProvider.SMTP.value,
nullable=False,
)
# =========================================================================
# SMTP Settings (used when provider=smtp)
# =========================================================================
smtp_host = Column(String(255), nullable=True)
smtp_port = Column(Integer, nullable=True, default=587)
smtp_username = Column(String(255), nullable=True)
smtp_password = Column(String(500), nullable=True) # Encrypted at rest
smtp_use_tls = Column(Boolean, default=True, nullable=False)
smtp_use_ssl = Column(Boolean, default=False, nullable=False) # For port 465
# =========================================================================
# SendGrid Settings (used when provider=sendgrid, Business+ tier)
# =========================================================================
sendgrid_api_key = Column(String(500), nullable=True) # Encrypted at rest
# =========================================================================
# Mailgun Settings (used when provider=mailgun, Business+ tier)
# =========================================================================
mailgun_api_key = Column(String(500), nullable=True) # Encrypted at rest
mailgun_domain = Column(String(255), nullable=True)
# =========================================================================
# Amazon SES Settings (used when provider=ses, Business+ tier)
# =========================================================================
ses_access_key_id = Column(String(100), nullable=True)
ses_secret_access_key = Column(String(500), nullable=True) # Encrypted at rest
ses_region = Column(String(50), nullable=True, default="eu-west-1")
# =========================================================================
# Status & Verification
# =========================================================================
is_configured = Column(Boolean, default=False, nullable=False) # Has complete config
is_verified = Column(Boolean, default=False, nullable=False) # Test email succeeded
last_verified_at = Column(DateTime(timezone=True), nullable=True)
verification_error = Column(Text, nullable=True) # Last verification error message
# =========================================================================
# Relationship
# =========================================================================
vendor = relationship("Vendor", back_populates="email_settings")
# =========================================================================
# Indexes
# =========================================================================
__table_args__ = (
Index("idx_vendor_email_settings_configured", "vendor_id", "is_configured"),
)
def __repr__(self) -> str:
return f"<VendorEmailSettings(vendor_id={self.vendor_id}, provider='{self.provider}', from='{self.from_email}')>"
# =========================================================================
# Helper Methods
# =========================================================================
def is_smtp_configured(self) -> bool:
"""Check if SMTP settings are complete."""
if self.provider != EmailProvider.SMTP.value:
return False
return bool(
self.smtp_host
and self.smtp_port
and self.smtp_username
and self.smtp_password
)
def is_sendgrid_configured(self) -> bool:
"""Check if SendGrid settings are complete."""
if self.provider != EmailProvider.SENDGRID.value:
return False
return bool(self.sendgrid_api_key)
def is_mailgun_configured(self) -> bool:
"""Check if Mailgun settings are complete."""
if self.provider != EmailProvider.MAILGUN.value:
return False
return bool(self.mailgun_api_key and self.mailgun_domain)
def is_ses_configured(self) -> bool:
"""Check if Amazon SES settings are complete."""
if self.provider != EmailProvider.SES.value:
return False
return bool(
self.ses_access_key_id
and self.ses_secret_access_key
and self.ses_region
)
def is_provider_configured(self) -> bool:
"""Check if the current provider is fully configured."""
provider_checks = {
EmailProvider.SMTP.value: self.is_smtp_configured,
EmailProvider.SENDGRID.value: self.is_sendgrid_configured,
EmailProvider.MAILGUN.value: self.is_mailgun_configured,
EmailProvider.SES.value: self.is_ses_configured,
}
check_fn = provider_checks.get(self.provider)
return check_fn() if check_fn else False
def is_fully_configured(self) -> bool:
"""Check if email settings are fully configured (identity + provider)."""
return bool(
self.from_email
and self.from_name
and self.is_provider_configured()
)
def update_configuration_status(self) -> None:
"""Update the is_configured flag based on current settings."""
self.is_configured = self.is_fully_configured()
def mark_verified(self) -> None:
"""Mark settings as verified (test email succeeded)."""
self.is_verified = True
self.last_verified_at = datetime.now(UTC)
self.verification_error = None
def mark_verification_failed(self, error: str) -> None:
"""Mark settings as verification failed."""
self.is_verified = False
self.verification_error = error
def requires_premium_tier(self) -> bool:
"""Check if current provider requires Business+ tier."""
return self.provider in [p.value for p in PREMIUM_EMAIL_PROVIDERS]
def to_dict(self) -> dict:
"""Convert to dictionary for API responses (excludes sensitive data)."""
return {
"id": self.id,
"vendor_id": self.vendor_id,
"from_email": self.from_email,
"from_name": self.from_name,
"reply_to_email": self.reply_to_email,
"signature_text": self.signature_text,
"signature_html": self.signature_html,
"provider": self.provider,
# SMTP (mask password)
"smtp_host": self.smtp_host,
"smtp_port": self.smtp_port,
"smtp_username": self.smtp_username,
"smtp_password_set": bool(self.smtp_password),
"smtp_use_tls": self.smtp_use_tls,
"smtp_use_ssl": self.smtp_use_ssl,
# SendGrid (mask API key)
"sendgrid_api_key_set": bool(self.sendgrid_api_key),
# Mailgun (mask API key)
"mailgun_api_key_set": bool(self.mailgun_api_key),
"mailgun_domain": self.mailgun_domain,
# SES (mask credentials)
"ses_access_key_id_set": bool(self.ses_access_key_id),
"ses_region": self.ses_region,
# Status
"is_configured": self.is_configured,
"is_verified": self.is_verified,
"last_verified_at": self.last_verified_at.isoformat() if self.last_verified_at else None,
"verification_error": self.verification_error,
"requires_premium_tier": self.requires_premium_tier(),
# Timestamps
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}

229
models/database/vendor_email_template.py
View File

@@ -1,229 +0,0 @@
# models/database/vendor_email_template.py
"""
Vendor email template override model.
Allows vendors to customize platform email templates with their own content.
Platform-only templates cannot be overridden (e.g., billing, subscription emails).
"""
from datetime import datetime
from sqlalchemy import (
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import Session, relationship
from app.core.database import Base
from .base import TimestampMixin
class VendorEmailTemplate(Base, TimestampMixin):
"""
Vendor-specific email template override.
Each vendor can customize email templates for their shop.
Overrides are per-template-code and per-language.
When sending emails:
1. Check if vendor has an override for the template+language
2. If yes, use vendor's version
3. If no, fall back to platform template
Platform-only templates (is_platform_only=True on EmailTemplate)
cannot be overridden.
"""
__tablename__ = "vendor_email_templates"
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
# Vendor relationship
vendor_id = Column(
Integer,
ForeignKey("vendors.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Template identification (references EmailTemplate.code, not FK)
template_code = Column(String(100), nullable=False, index=True)
language = Column(String(5), nullable=False, default="en")
# Optional custom name (if None, uses platform template name)
name = Column(String(255), nullable=True)
# Email content
subject = Column(String(500), nullable=False)
body_html = Column(Text, nullable=False)
body_text = Column(Text, nullable=True)
# Status
is_active = Column(Boolean, default=True, nullable=False)
# Relationships
vendor = relationship("Vendor", back_populates="email_templates")
# Unique constraint: one override per vendor+template+language
__table_args__ = (
UniqueConstraint(
"vendor_id",
"template_code",
"language",
name="uq_vendor_email_template_code_language",
),
{"sqlite_autoincrement": True},
)
def __repr__(self):
return (
f"<VendorEmailTemplate("
f"vendor_id={self.vendor_id}, "
f"code='{self.template_code}', "
f"language='{self.language}')>"
)
@classmethod
def get_override(
cls,
db: Session,
vendor_id: int,
template_code: str,
language: str,
) -> "VendorEmailTemplate | None":
"""
Get vendor's template override if it exists.
Args:
db: Database session
vendor_id: Vendor ID
template_code: Template code to look up
language: Language code (en, fr, de, lb)
Returns:
VendorEmailTemplate if override exists, None otherwise
"""
return (
db.query(cls)
.filter(
cls.vendor_id == vendor_id,
cls.template_code == template_code,
cls.language == language,
cls.is_active == True, # noqa: E712
)
.first()
)
@classmethod
def get_all_overrides_for_vendor(
cls,
db: Session,
vendor_id: int,
) -> list["VendorEmailTemplate"]:
"""
Get all template overrides for a vendor.
Args:
db: Database session
vendor_id: Vendor ID
Returns:
List of VendorEmailTemplate objects
"""
return (
db.query(cls)
.filter(
cls.vendor_id == vendor_id,
cls.is_active == True, # noqa: E712
)
.order_by(cls.template_code, cls.language)
.all()
)
@classmethod
def create_or_update(
cls,
db: Session,
vendor_id: int,
template_code: str,
language: str,
subject: str,
body_html: str,
body_text: str | None = None,
name: str | None = None,
) -> "VendorEmailTemplate":
"""
Create or update a vendor email template override.
Args:
db: Database session
vendor_id: Vendor ID
template_code: Template code
language: Language code
subject: Email subject
body_html: HTML body content
body_text: Optional plain text body
name: Optional custom name
Returns:
Created or updated VendorEmailTemplate
"""
existing = cls.get_override(db, vendor_id, template_code, language)
if existing:
existing.subject = subject
existing.body_html = body_html
existing.body_text = body_text
existing.name = name
existing.updated_at = datetime.utcnow()
return existing
new_template = cls(
vendor_id=vendor_id,
template_code=template_code,
language=language,
subject=subject,
body_html=body_html,
body_text=body_text,
name=name,
)
db.add(new_template)
return new_template
@classmethod
def delete_override(
cls,
db: Session,
vendor_id: int,
template_code: str,
language: str,
) -> bool:
"""
Delete a vendor's template override (revert to platform default).
Args:
db: Database session
vendor_id: Vendor ID
template_code: Template code
language: Language code
Returns:
True if deleted, False if not found
"""
deleted = (
db.query(cls)
.filter(
cls.vendor_id == vendor_id,
cls.template_code == template_code,
cls.language == language,
)
.delete()
)
return deleted > 0

189
models/database/vendor_platform.py
View File

@@ -1,189 +0,0 @@
# 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})>"
)

136
models/database/vendor_theme.py
View File

@@ -1,136 +0,0 @@
# models/database/vendor_theme.py
"""
Vendor Theme Configuration Model
Allows each vendor to customize their shop's appearance
"""
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship
from app.core.database import Base
from models.database.base import TimestampMixin
class VendorTheme(Base, TimestampMixin):
"""
Stores theme configuration for each vendor's shop.
Each vendor can have ONE active theme:
- Custom colors (primary, secondary, accent)
- Custom fonts
- Custom logo and favicon
- Custom CSS overrides
- Layout preferences
Theme presets available: default, modern, classic, minimal, vibrant
"""
__tablename__ = "vendor_themes"
id = Column(Integer, primary_key=True, index=True)
vendor_id = Column(
Integer,
ForeignKey("vendors.id", ondelete="CASCADE"),
nullable=False,
unique=True, # ONE vendor = ONE theme
)
# Basic Theme Settings
theme_name = Column(
String(100), default="default"
) # default, modern, classic, minimal, vibrant
is_active = Column(Boolean, default=True)
# Color Scheme (JSON for flexibility)
colors = Column(
JSON,
default={
"primary": "#6366f1", # Indigo
"secondary": "#8b5cf6", # Purple
"accent": "#ec4899", # Pink
"background": "#ffffff", # White
"text": "#1f2937", # Gray-800
"border": "#e5e7eb", # Gray-200
},
)
# Typography
font_family_heading = Column(String(100), default="Inter, sans-serif")
font_family_body = Column(String(100), default="Inter, sans-serif")
# Branding Assets
logo_url = Column(String(500), nullable=True) # Path to vendor logo
logo_dark_url = Column(String(500), nullable=True) # Dark mode logo
favicon_url = Column(String(500), nullable=True) # Favicon
banner_url = Column(String(500), nullable=True) # Homepage banner
# Layout Preferences
layout_style = Column(String(50), default="grid") # grid, list, masonry
header_style = Column(String(50), default="fixed") # fixed, static, transparent
product_card_style = Column(
String(50), default="modern"
) # modern, classic, minimal
# Custom CSS (for advanced customization)
custom_css = Column(Text, nullable=True)
# Social Media Links
social_links = Column(JSON, default={}) # {facebook: "url", instagram: "url", etc.}
# SEO & Meta
meta_title_template = Column(
String(200), nullable=True
) # e.g., "{product_name} - {shop_name}"
meta_description = Column(Text, nullable=True)
# Relationships - FIXED: back_populates must match the relationship name in Vendor model
vendor = relationship("Vendor", back_populates="vendor_theme")
def __repr__(self):
return (
f"<VendorTheme(vendor_id={self.vendor_id}, theme_name='{self.theme_name}')>"
)
@property
def primary_color(self):
"""Get primary color from JSON"""
return self.colors.get("primary", "#6366f1")
@property
def css_variables(self):
"""Generate CSS custom properties from theme config"""
return {
"--color-primary": self.colors.get("primary", "#6366f1"),
"--color-secondary": self.colors.get("secondary", "#8b5cf6"),
"--color-accent": self.colors.get("accent", "#ec4899"),
"--color-background": self.colors.get("background", "#ffffff"),
"--color-text": self.colors.get("text", "#1f2937"),
"--color-border": self.colors.get("border", "#e5e7eb"),
"--font-heading": self.font_family_heading,
"--font-body": self.font_family_body,
}
def to_dict(self):
"""Convert theme to dictionary for template rendering"""
return {
"theme_name": self.theme_name,
"colors": self.colors,
"fonts": {
"heading": self.font_family_heading,
"body": self.font_family_body,
},
"branding": {
"logo": self.logo_url,
"logo_dark": self.logo_dark_url,
"favicon": self.favicon_url,
"banner": self.banner_url,
},
"layout": {
"style": self.layout_style,
"header": self.header_style,
"product_card": self.product_card_style,
},
"social_links": self.social_links,
"custom_css": self.custom_css,
"css_variables": self.css_variables,
}

21
models/schema/__init__.py
View File

@@ -1,23 +1,30 @@
# models/schema/__init__.py
"""API models package - Pydantic models for request/response validation.
"""API models package - Base classes only.
Note: Many schemas have been migrated to their respective modules:
This package provides the base infrastructure for Pydantic schemas:
- BaseModel configuration
- Common response patterns
- Auth schemas (cross-cutting)
IMPORTANT: Domain schemas have been migrated to their respective modules:
- Tenancy schemas: app.modules.tenancy.schemas
- CMS schemas: app.modules.cms.schemas
- Messaging schemas: app.modules.messaging.schemas
- Customer schemas: app.modules.customers.schemas
- Order schemas: app.modules.orders.schemas
- Inventory schemas: app.modules.inventory.schemas
- Message schemas: app.modules.messaging.schemas
- Cart schemas: app.modules.cart.schemas
- Marketplace schemas: app.modules.marketplace.schemas
- Catalog/Product schemas: app.modules.catalog.schemas
- Payment schemas: app.modules.payments.schemas
Import schemas from their canonical module locations instead of this package.
"""
# Import API model modules that remain in legacy location
# Infrastructure schemas that remain here
from . import (
auth,
base,
email,
vendor,
)
# Common imports for convenience
@@ -26,6 +33,4 @@ from .base import * # Base Pydantic models
__all__ = [
"base",
"auth",
"email",
"vendor",
]

590
models/schema/admin.py
View File

@@ -1,590 +0,0 @@
# models/schema/admin.py
"""
Admin-specific Pydantic schemas for API validation and responses.
This module provides schemas for:
- Admin audit logs
- Admin notifications
- Platform settings
- Platform alerts
- Bulk operations
- System health checks
"""
from datetime import datetime
from typing import Any
from pydantic import BaseModel, Field, field_validator
# ============================================================================
# ADMIN AUDIT LOG SCHEMAS
# ============================================================================
class AdminAuditLogResponse(BaseModel):
"""Response model for admin audit logs."""
id: int
admin_user_id: int
admin_username: str | None = None
action: str
target_type: str
target_id: str
details: dict[str, Any] | None = None
ip_address: str | None = None
user_agent: str | None = None
request_id: str | None = None
created_at: datetime
model_config = {"from_attributes": True}
class AdminAuditLogFilters(BaseModel):
"""Filters for querying audit logs."""
admin_user_id: int | None = None
action: str | None = None
target_type: str | None = None
date_from: datetime | None = None
date_to: datetime | None = None
skip: int = Field(0, ge=0)
limit: int = Field(100, ge=1, le=1000)
class AdminAuditLogListResponse(BaseModel):
"""Paginated list of audit logs."""
logs: list[AdminAuditLogResponse]
total: int
skip: int
limit: int
# ============================================================================
# ADMIN NOTIFICATION SCHEMAS
# ============================================================================
class AdminNotificationCreate(BaseModel):
"""Create admin notification."""
type: str = Field(..., max_length=50, description="Notification type")
priority: str = Field(default="normal", description="Priority level")
title: str = Field(..., max_length=200)
message: str = Field(..., description="Notification message")
action_required: bool = Field(default=False)
action_url: str | None = Field(None, max_length=500)
metadata: dict[str, Any] | None = None
@field_validator("priority")
@classmethod
def validate_priority(cls, v):
allowed = ["low", "normal", "high", "critical"]
if v not in allowed:
raise ValueError(f"Priority must be one of: {', '.join(allowed)}")
return v
class AdminNotificationResponse(BaseModel):
"""Admin notification response."""
id: int
type: str
priority: str
title: str
message: str
is_read: bool
read_at: datetime | None = None
read_by_user_id: int | None = None
action_required: bool
action_url: str | None = None
metadata: dict[str, Any] | None = None
created_at: datetime
model_config = {"from_attributes": True}
class AdminNotificationUpdate(BaseModel):
"""Mark notification as read."""
is_read: bool = True
class AdminNotificationListResponse(BaseModel):
"""Paginated list of notifications."""
notifications: list[AdminNotificationResponse]
total: int
unread_count: int
skip: int
limit: int
# ============================================================================
# ADMIN SETTINGS SCHEMAS
# ============================================================================
class AdminSettingCreate(BaseModel):
"""Create or update admin setting."""
key: str = Field(..., max_length=100, description="Unique setting key")
value: str = Field(..., description="Setting value")
value_type: str = Field(default="string", description="Data type")
category: str | None = Field(None, max_length=50)
description: str | None = None
is_encrypted: bool = Field(default=False)
is_public: bool = Field(default=False, description="Can be exposed to frontend")
@field_validator("value_type")
@classmethod
def validate_value_type(cls, v):
allowed = ["string", "integer", "boolean", "json", "float"]
if v not in allowed:
raise ValueError(f"Value type must be one of: {', '.join(allowed)}")
return v
@field_validator("key")
@classmethod
def validate_key_format(cls, v):
# Setting keys should be lowercase with underscores
if not v.replace("_", "").isalnum():
raise ValueError(
"Setting key must contain only letters, numbers, and underscores"
)
return v.lower()
class AdminSettingResponse(BaseModel):
"""Admin setting response."""
id: int
key: str
value: str
value_type: str
category: str | None = None
description: str | None = None
is_encrypted: bool
is_public: bool
last_modified_by_user_id: int | None = None
updated_at: datetime
model_config = {"from_attributes": True}
class AdminSettingDefaultResponse(BaseModel):
"""Response when returning a default value for non-existent setting."""
key: str
value: str
exists: bool = False
class AdminSettingUpdate(BaseModel):
"""Update admin setting value."""
value: str
description: str | None = None
class AdminSettingListResponse(BaseModel):
"""List of settings by category."""
settings: list[AdminSettingResponse]
total: int
category: str | None = None
# ============================================================================
# DISPLAY SETTINGS SCHEMAS
# ============================================================================
class RowsPerPageResponse(BaseModel):
"""Response for rows per page setting."""
rows_per_page: int
class RowsPerPageUpdateResponse(BaseModel):
"""Response after updating rows per page."""
rows_per_page: int
message: str
class PublicDisplaySettingsResponse(BaseModel):
"""Public display settings (no auth required)."""
rows_per_page: int
# ============================================================================
# PLATFORM ALERT SCHEMAS
# ============================================================================
class PlatformAlertCreate(BaseModel):
"""Create platform alert."""
alert_type: str = Field(..., max_length=50)
severity: str = Field(..., description="Alert severity")
title: str = Field(..., max_length=200)
description: str | None = None
affected_vendors: list[int] | None = None
affected_systems: list[str] | None = None
auto_generated: bool = Field(default=True)
@field_validator("severity")
@classmethod
def validate_severity(cls, v):
allowed = ["info", "warning", "error", "critical"]
if v not in allowed:
raise ValueError(f"Severity must be one of: {', '.join(allowed)}")
return v
@field_validator("alert_type")
@classmethod
def validate_alert_type(cls, v):
allowed = [
"security",
"performance",
"capacity",
"integration",
"database",
"system",
]
if v not in allowed:
raise ValueError(f"Alert type must be one of: {', '.join(allowed)}")
return v
class PlatformAlertResponse(BaseModel):
"""Platform alert response."""
id: int
alert_type: str
severity: str
title: str
description: str | None = None
affected_vendors: list[int] | None = None
affected_systems: list[str] | None = None
is_resolved: bool
resolved_at: datetime | None = None
resolved_by_user_id: int | None = None
resolution_notes: str | None = None
auto_generated: bool
occurrence_count: int
first_occurred_at: datetime
last_occurred_at: datetime
created_at: datetime
model_config = {"from_attributes": True}
class PlatformAlertResolve(BaseModel):
"""Resolve platform alert."""
is_resolved: bool = True
resolution_notes: str | None = None
class PlatformAlertListResponse(BaseModel):
"""Paginated list of platform alerts."""
alerts: list[PlatformAlertResponse]
total: int
active_count: int
critical_count: int
skip: int
limit: int
# ============================================================================
# BULK OPERATION SCHEMAS
# ============================================================================
class BulkVendorAction(BaseModel):
"""Bulk actions on vendors."""
vendor_ids: list[int] = Field(..., min_length=1, max_length=100)
action: str = Field(..., description="Action to perform")
confirm: bool = Field(default=False, description="Required for destructive actions")
reason: str | None = Field(None, description="Reason for bulk action")
@field_validator("action")
@classmethod
def validate_action(cls, v):
allowed = ["activate", "deactivate", "verify", "unverify", "delete"]
if v not in allowed:
raise ValueError(f"Action must be one of: {', '.join(allowed)}")
return v
class BulkVendorActionResponse(BaseModel):
"""Response for bulk vendor actions."""
successful: list[int]
failed: dict[int, str] # vendor_id -> error_message
total_processed: int
action_performed: str
message: str
class BulkUserAction(BaseModel):
"""Bulk actions on users."""
user_ids: list[int] = Field(..., min_length=1, max_length=100)
action: str = Field(..., description="Action to perform")
confirm: bool = Field(default=False)
reason: str | None = None
@field_validator("action")
@classmethod
def validate_action(cls, v):
allowed = ["activate", "deactivate", "delete"]
if v not in allowed:
raise ValueError(f"Action must be one of: {', '.join(allowed)}")
return v
class BulkUserActionResponse(BaseModel):
"""Response for bulk user actions."""
successful: list[int]
failed: dict[int, str]
total_processed: int
action_performed: str
message: str
# ============================================================================
# ADMIN DASHBOARD SCHEMAS
# ============================================================================
class AdminDashboardStats(BaseModel):
"""Comprehensive admin dashboard statistics."""
platform: dict[str, Any]
users: dict[str, Any]
vendors: dict[str, Any]
products: dict[str, Any]
orders: dict[str, Any]
imports: dict[str, Any]
recent_vendors: list[dict[str, Any]]
recent_imports: list[dict[str, Any]]
unread_notifications: int
active_alerts: int
critical_alerts: int
# ============================================================================
# SYSTEM HEALTH SCHEMAS
# ============================================================================
class ComponentHealthStatus(BaseModel):
"""Health status for a system component."""
status: str # healthy, degraded, unhealthy
response_time_ms: float | None = None
error_message: str | None = None
last_checked: datetime
details: dict[str, Any] | None = None
class SystemHealthResponse(BaseModel):
"""System health check response."""
overall_status: str # healthy, degraded, critical
database: ComponentHealthStatus
redis: ComponentHealthStatus
celery: ComponentHealthStatus
storage: ComponentHealthStatus
api_response_time_ms: float
uptime_seconds: int
timestamp: datetime
# ============================================================================
# ADMIN SESSION SCHEMAS
# ============================================================================
class AdminSessionResponse(BaseModel):
"""Admin session information."""
id: int
admin_user_id: int
admin_username: str | None = None
ip_address: str
user_agent: str | None = None
login_at: datetime
last_activity_at: datetime
logout_at: datetime | None = None
is_active: bool
logout_reason: str | None = None
model_config = {"from_attributes": True}
class AdminSessionListResponse(BaseModel):
"""List of admin sessions."""
sessions: list[AdminSessionResponse]
total: int
active_count: int
# ============================================================================
# APPLICATION LOGS SCHEMAS
# ============================================================================
class ApplicationLogResponse(BaseModel):
"""Application log entry response."""
id: int
timestamp: datetime
level: str
logger_name: str
module: str | None = None
function_name: str | None = None
line_number: int | None = None
message: str
exception_type: str | None = None
exception_message: str | None = None
stack_trace: str | None = None
request_id: str | None = None
user_id: int | None = None
vendor_id: int | None = None
context: dict[str, Any] | None = None
created_at: datetime
model_config = {"from_attributes": True}
class ApplicationLogFilters(BaseModel):
"""Filters for querying application logs."""
level: str | None = Field(None, description="Filter by log level")
logger_name: str | None = Field(None, description="Filter by logger name")
module: str | None = Field(None, description="Filter by module")
user_id: int | None = Field(None, description="Filter by user ID")
vendor_id: int | None = Field(None, description="Filter by vendor ID")
date_from: datetime | None = Field(None, description="Start date")
date_to: datetime | None = Field(None, description="End date")
search: str | None = Field(None, description="Search in message")
skip: int = Field(0, ge=0)
limit: int = Field(100, ge=1, le=1000)
class ApplicationLogListResponse(BaseModel):
"""Paginated list of application logs."""
logs: list[ApplicationLogResponse]
total: int
skip: int
limit: int
class LogStatistics(BaseModel):
"""Statistics about application logs."""
total_count: int
warning_count: int
error_count: int
critical_count: int
by_level: dict[str, int]
by_module: dict[str, int]
recent_errors: list[ApplicationLogResponse]
# ============================================================================
# LOG SETTINGS SCHEMAS
# ============================================================================
class LogSettingsResponse(BaseModel):
"""Log configuration settings."""
log_level: str
log_file_max_size_mb: int
log_file_backup_count: int
db_log_retention_days: int
file_logging_enabled: bool
db_logging_enabled: bool
class LogSettingsUpdate(BaseModel):
"""Update log settings."""
log_level: str | None = Field(
None, description="Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL"
)
log_file_max_size_mb: int | None = Field(
None, ge=1, le=1000, description="Max log file size in MB"
)
log_file_backup_count: int | None = Field(
None, ge=0, le=50, description="Number of backup files to keep"
)
db_log_retention_days: int | None = Field(
None, ge=1, le=365, description="Days to retain logs in database"
)
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v):
if v is not None:
allowed = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
if v.upper() not in allowed:
raise ValueError(f"Log level must be one of: {', '.join(allowed)}")
return v.upper()
return v
class FileLogResponse(BaseModel):
"""File log content response."""
filename: str
size_bytes: int
last_modified: datetime
lines: list[str]
total_lines: int
class LogFileInfo(BaseModel):
"""Log file info for listing."""
filename: str
size_bytes: int
last_modified: datetime
class LogFileListResponse(BaseModel):
"""Response for listing log files."""
files: list[LogFileInfo]
class LogDeleteResponse(BaseModel):
"""Response for log deletion."""
message: str
class LogCleanupResponse(BaseModel):
"""Response for log cleanup operation."""
message: str
deleted_count: int
class LogSettingsUpdateResponse(BaseModel):
"""Response for log settings update."""
message: str
updated_fields: list[str]
note: str | None = None

216
models/schema/company.py
View File

@@ -1,216 +0,0 @@
# models/schema/company.py
"""
Pydantic schemas for Company model.
These schemas are used for API request/response validation and serialization.
"""
from datetime import datetime
from typing import Any
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
class CompanyBase(BaseModel):
"""Base schema for company with common fields."""
name: str = Field(..., min_length=2, max_length=200, description="Company name")
description: str | None = Field(None, description="Company description")
contact_email: EmailStr = Field(..., description="Business contact email")
contact_phone: str | None = Field(None, description="Business phone number")
website: str | None = Field(None, description="Company website URL")
business_address: str | None = Field(None, description="Physical business address")
tax_number: str | None = Field(None, description="Tax/VAT registration number")
@field_validator("contact_email")
@classmethod
def normalize_email(cls, v):
"""Normalize email to lowercase."""
return v.lower() if v else v
class CompanyCreate(CompanyBase):
"""
Schema for creating a new company.
Requires owner_email to create the associated owner user account.
"""
owner_email: EmailStr = Field(
..., description="Email for the company owner account"
)
@field_validator("owner_email")
@classmethod
def normalize_owner_email(cls, v):
"""Normalize owner email to lowercase."""
return v.lower() if v else v
model_config = ConfigDict(from_attributes=True)
class CompanyUpdate(BaseModel):
"""
Schema for updating company information.
All fields are optional to support partial updates.
"""
name: str | None = Field(None, min_length=2, max_length=200)
description: str | None = None
contact_email: EmailStr | None = None
contact_phone: str | None = None
website: str | None = None
business_address: str | None = None
tax_number: str | None = None
# Status (Admin only)
is_active: bool | None = None
is_verified: bool | None = None
@field_validator("contact_email")
@classmethod
def normalize_email(cls, v):
"""Normalize email to lowercase."""
return v.lower() if v else v
model_config = ConfigDict(from_attributes=True)
class CompanyResponse(BaseModel):
"""Standard schema for company response data."""
model_config = ConfigDict(from_attributes=True)
id: int
name: str
description: str | None
# Owner information
owner_user_id: int
# Contact Information
contact_email: str
contact_phone: str | None
website: str | None
# Business Information
business_address: str | None
tax_number: str | None
# Status Flags
is_active: bool
is_verified: bool
# Timestamps
created_at: str
updated_at: str
class CompanyDetailResponse(CompanyResponse):
"""
Detailed company response including vendor count and owner details.
Used for company detail pages and admin views.
"""
# Owner details (from related User)
owner_email: str | None = Field(None, description="Owner's email address")
owner_username: str | None = Field(None, description="Owner's username")
# Vendor statistics
vendor_count: int = Field(0, description="Number of vendors under this company")
active_vendor_count: int = Field(
0, description="Number of active vendors under this company"
)
# Vendors list (optional, for detail view)
vendors: list | None = Field(None, description="List of vendors under this company")
class CompanyListResponse(BaseModel):
"""Schema for paginated company list."""
companies: list[CompanyResponse]
total: int
skip: int
limit: int
class CompanyCreateResponse(BaseModel):
"""
Response after creating a company with owner account.
Includes temporary password for the owner (shown only once).
"""
company: CompanyResponse
owner_user_id: int
owner_username: str
owner_email: str
temporary_password: str = Field(
..., description="Temporary password for owner (SHOWN ONLY ONCE)"
)
login_url: str | None = Field(None, description="URL for company owner to login")
class CompanySummary(BaseModel):
"""Lightweight company summary for dropdowns and quick references."""
model_config = ConfigDict(from_attributes=True)
id: int
name: str
is_active: bool
is_verified: bool
vendor_count: int = 0
class CompanyTransferOwnership(BaseModel):
"""
Schema for transferring company ownership to another user.
This is a critical operation that requires:
- Confirmation flag
- Reason for audit trail (optional)
"""
new_owner_user_id: int = Field(
..., description="ID of the user who will become the new owner", gt=0
)
confirm_transfer: bool = Field(
..., description="Must be true to confirm ownership transfer"
)
transfer_reason: str | None = Field(
None,
max_length=500,
description="Reason for ownership transfer (for audit logs)",
)
@field_validator("confirm_transfer")
@classmethod
def validate_confirmation(cls, v):
"""Ensure confirmation is explicitly true."""
if not v:
raise ValueError("Ownership transfer requires explicit confirmation")
return v
class CompanyTransferOwnershipResponse(BaseModel):
"""Response after successful ownership transfer."""
message: str
company_id: int
company_name: str
old_owner: dict[str, Any] = Field(
..., description="Information about the previous owner"
)
new_owner: dict[str, Any] = Field(
..., description="Information about the new owner"
)
transferred_at: datetime
transfer_reason: str | None

247
models/schema/email.py
View File

@@ -1,247 +0,0 @@
# models/schema/email.py
"""
Email template Pydantic schemas for API responses and requests.
Provides schemas for:
- EmailTemplate: Platform email templates
- VendorEmailTemplate: Vendor-specific template overrides
"""
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class EmailTemplateBase(BaseModel):
"""Base schema for email templates."""
code: str = Field(..., description="Template code (e.g., 'password_reset')")
language: str = Field(default="en", description="Language code (en, fr, de, lb)")
name: str = Field(..., description="Human-readable template name")
description: str | None = Field(None, description="Template description")
category: str = Field(..., description="Template category (auth, orders, billing, etc.)")
subject: str = Field(..., description="Email subject (supports Jinja2 variables)")
body_html: str = Field(..., description="HTML email body")
body_text: str | None = Field(None, description="Plain text fallback")
variables: list[str] = Field(default_factory=list, description="Available variables")
class EmailTemplateCreate(EmailTemplateBase):
"""Schema for creating an email template."""
required_variables: list[str] = Field(
default_factory=list, description="Required variables"
)
is_platform_only: bool = Field(
default=False, description="Cannot be overridden by vendors"
)
class EmailTemplateUpdate(BaseModel):
"""Schema for updating an email template."""
name: str | None = Field(None, description="Human-readable template name")
description: str | None = Field(None, description="Template description")
subject: str | None = Field(None, description="Email subject")
body_html: str | None = Field(None, description="HTML email body")
body_text: str | None = Field(None, description="Plain text fallback")
variables: list[str] | None = Field(None, description="Available variables")
required_variables: list[str] | None = Field(None, description="Required variables")
is_active: bool | None = Field(None, description="Template active status")
class EmailTemplateResponse(BaseModel):
"""Schema for email template API response."""
model_config = ConfigDict(from_attributes=True)
id: int
code: str
language: str
name: str
description: str | None
category: str
subject: str
body_html: str
body_text: str | None
variables: list[str] = Field(default_factory=list)
required_variables: list[str] = Field(default_factory=list)
is_active: bool
is_platform_only: bool
created_at: datetime
updated_at: datetime
@classmethod
def from_db(cls, template) -> "EmailTemplateResponse":
"""Create response from database model."""
return cls(
id=template.id,
code=template.code,
language=template.language,
name=template.name,
description=template.description,
category=template.category,
subject=template.subject,
body_html=template.body_html,
body_text=template.body_text,
variables=template.variables_list,
required_variables=template.required_variables_list,
is_active=template.is_active,
is_platform_only=template.is_platform_only,
created_at=template.created_at,
updated_at=template.updated_at,
)
class EmailTemplateSummary(BaseModel):
"""Summary schema for template list views."""
model_config = ConfigDict(from_attributes=True)
id: int
code: str
name: str
category: str
languages: list[str] = Field(default_factory=list)
is_platform_only: bool
is_active: bool
@classmethod
def from_db_list(cls, templates: list) -> list["EmailTemplateSummary"]:
"""
Create summaries from database models, grouping by code.
Args:
templates: List of EmailTemplate models
Returns:
List of EmailTemplateSummary grouped by template code
"""
# Group templates by code
by_code: dict[str, list] = {}
for t in templates:
if t.code not in by_code:
by_code[t.code] = []
by_code[t.code].append(t)
summaries = []
for code, group in by_code.items():
first = group[0]
summaries.append(
cls(
id=first.id,
code=code,
name=first.name,
category=first.category,
languages=[t.language for t in group],
is_platform_only=first.is_platform_only,
is_active=first.is_active,
)
)
return summaries
# Vendor Email Template Schemas
class VendorEmailTemplateCreate(BaseModel):
"""Schema for creating a vendor email template override."""
template_code: str = Field(..., description="Template code to override")
language: str = Field(default="en", description="Language code")
name: str | None = Field(None, description="Custom name (uses platform default if None)")
subject: str = Field(..., description="Custom email subject")
body_html: str = Field(..., description="Custom HTML body")
body_text: str | None = Field(None, description="Custom plain text body")
class VendorEmailTemplateUpdate(BaseModel):
"""Schema for updating a vendor email template override."""
name: str | None = Field(None, description="Custom name")
subject: str | None = Field(None, description="Custom email subject")
body_html: str | None = Field(None, description="Custom HTML body")
body_text: str | None = Field(None, description="Custom plain text body")
is_active: bool | None = Field(None, description="Override active status")
class VendorEmailTemplateResponse(BaseModel):
"""Schema for vendor email template override API response."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_id: int
template_code: str
language: str
name: str | None
subject: str
body_html: str
body_text: str | None
is_active: bool
created_at: datetime
updated_at: datetime
class EmailTemplateWithOverrideStatus(BaseModel):
"""
Schema showing template with vendor override status.
Used in vendor UI to show which templates have been customized.
"""
model_config = ConfigDict(from_attributes=True)
code: str
name: str
category: str
languages: list[str]
is_platform_only: bool
has_override: bool = Field(
default=False, description="Whether vendor has customized this template"
)
override_languages: list[str] = Field(
default_factory=list,
description="Languages with vendor overrides",
)
# Email Preview/Test Schemas
class EmailPreviewRequest(BaseModel):
"""Schema for requesting an email preview."""
template_code: str = Field(..., description="Template code")
language: str = Field(default="en", description="Language code")
variables: dict[str, str] = Field(
default_factory=dict, description="Variables to inject"
)
class EmailPreviewResponse(BaseModel):
"""Schema for email preview response."""
subject: str
body_html: str
body_text: str | None
class EmailTestRequest(BaseModel):
"""Schema for sending a test email."""
template_code: str = Field(..., description="Template code")
language: str = Field(default="en", description="Language code")
to_email: str = Field(..., description="Recipient email address")
variables: dict[str, str] = Field(
default_factory=dict, description="Variables to inject"
)
class EmailTestResponse(BaseModel):
"""Schema for test email response."""
success: bool
message: str
email_log_id: int | None = None

46
models/schema/image.py
View File

@@ -1,46 +0,0 @@
# models/schema/image.py
"""
Pydantic schemas for image operations.
"""
from pydantic import BaseModel
class ImageUrls(BaseModel):
"""URLs for image variants."""
original: str
medium: str | None = None # 800px variant
thumb: str | None = None # 200px variant
# Allow arbitrary keys for flexibility
class Config:
extra = "allow"
class ImageUploadResponse(BaseModel):
"""Response from image upload."""
success: bool
image: dict | None = None
error: str | None = None
class ImageDeleteResponse(BaseModel):
"""Response from image deletion."""
success: bool
message: str
class ImageStorageStats(BaseModel):
"""Image storage statistics."""
total_files: int
total_size_bytes: int
total_size_mb: float
total_size_gb: float
directory_count: int
max_files_per_dir: int
avg_files_per_dir: float
products_estimated: int

198
models/schema/media.py
View File

@@ -1,198 +0,0 @@
# models/schema/media.py
"""
Media/file management Pydantic schemas for API validation and responses.
This module provides schemas for:
- Media library listing
- File upload responses
- Media metadata operations
- Media usage tracking
"""
from datetime import datetime
from typing import Any
from pydantic import BaseModel, Field
# ============================================================================
# SHARED RESPONSE SCHEMAS
# ============================================================================
class MessageResponse(BaseModel):
"""Generic message response for simple operations."""
message: str
# ============================================================================
# MEDIA ITEM SCHEMAS
# ============================================================================
class MediaItemResponse(BaseModel):
"""Single media item response."""
id: int
filename: str
original_filename: str | None = None
file_url: str
url: str | None = None # Alias for file_url for JS compatibility
thumbnail_url: str | None = None
media_type: str # image, video, document
mime_type: str | None = None
file_size: int | None = None # bytes
width: int | None = None # for images/videos
height: int | None = None # for images/videos
alt_text: str | None = None
description: str | None = None
folder: str | None = None
extra_metadata: dict[str, Any] | None = None
created_at: datetime
updated_at: datetime | None = None
model_config = {"from_attributes": True}
def model_post_init(self, __context: Any) -> None:
"""Set url from file_url if not provided."""
if self.url is None:
object.__setattr__(self, "url", self.file_url)
class MediaListResponse(BaseModel):
"""Paginated list of media items."""
media: list[MediaItemResponse] = []
total: int = 0
skip: int = 0
limit: int = 100
message: str | None = None
# ============================================================================
# UPLOAD RESPONSE SCHEMAS
# ============================================================================
class MediaUploadResponse(BaseModel):
"""Response for single file upload."""
success: bool = True
message: str | None = None
media: MediaItemResponse | None = None
# Legacy fields for backwards compatibility
id: int | None = None
file_url: str | None = None
thumbnail_url: str | None = None
filename: str | None = None
file_size: int | None = None
media_type: str | None = None
class UploadedFileInfo(BaseModel):
"""Information about a successfully uploaded file."""
id: int
filename: str
file_url: str
thumbnail_url: str | None = None
class FailedFileInfo(BaseModel):
"""Information about a failed file upload."""
filename: str
error: str
class MultipleUploadResponse(BaseModel):
"""Response for multiple file upload."""
uploaded_files: list[UploadedFileInfo] = []
failed_files: list[FailedFileInfo] = []
total_uploaded: int = 0
total_failed: int = 0
message: str | None = None
# ============================================================================
# MEDIA DETAIL SCHEMAS
# ============================================================================
class MediaDetailResponse(BaseModel):
"""Detailed media item response with usage info."""
id: int | None = None
filename: str | None = None
original_filename: str | None = None
file_url: str | None = None
thumbnail_url: str | None = None
media_type: str | None = None
mime_type: str | None = None
file_size: int | None = None
width: int | None = None
height: int | None = None
alt_text: str | None = None
description: str | None = None
folder: str | None = None
extra_metadata: dict[str, Any] | None = None
created_at: datetime | None = None
updated_at: datetime | None = None
message: str | None = None
model_config = {"from_attributes": True}
# ============================================================================
# MEDIA UPDATE SCHEMAS
# ============================================================================
class MediaMetadataUpdate(BaseModel):
"""Request model for updating media metadata."""
filename: str | None = Field(None, max_length=255)
alt_text: str | None = Field(None, max_length=500)
description: str | None = None
folder: str | None = Field(None, max_length=100)
metadata: dict[str, Any] | None = None # Named 'metadata' in API, stored as 'extra_metadata'
# ============================================================================
# MEDIA USAGE SCHEMAS
# ============================================================================
class ProductUsageInfo(BaseModel):
"""Information about product using this media."""
product_id: int
product_name: str
usage_type: str # main_image, gallery, variant, etc.
class MediaUsageResponse(BaseModel):
"""Response showing where media is being used."""
media_id: int | None = None
products: list[ProductUsageInfo] = []
other_usage: list[dict[str, Any]] = []
total_usage_count: int = 0
message: str | None = None
# ============================================================================
# MEDIA OPTIMIZATION SCHEMAS
# ============================================================================
class OptimizationResultResponse(BaseModel):
"""Response for media optimization operation."""
media_id: int | None = None
original_size: int | None = None
optimized_size: int | None = None
savings_percent: float | None = None
optimized_url: str | None = None
message: str | None = None

293
models/schema/team.py
View File

@@ -1,293 +0,0 @@
# models/schema/team.py
"""
Pydantic schemas for vendor team management.
This module defines request/response schemas for:
- Team member listing
- Team member invitation
- Team member updates
- Role management
"""
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, field_validator
# ============================================================================
# Role Schemas
# ============================================================================
class RoleBase(BaseModel):
"""Base role schema."""
name: str = Field(..., min_length=1, max_length=100, description="Role name")
permissions: list[str] = Field(
default_factory=list, description="List of permission strings"
)
class RoleCreate(RoleBase):
"""Schema for creating a role."""
class RoleUpdate(BaseModel):
"""Schema for updating a role."""
name: str | None = Field(None, min_length=1, max_length=100)
permissions: list[str] | None = None
class RoleResponse(RoleBase):
"""Schema for role response."""
id: int
vendor_id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True # Pydantic v2 (use orm_mode = True for v1)
class RoleListResponse(BaseModel):
"""Schema for role list response."""
roles: list[RoleResponse]
total: int
# ============================================================================
# Team Member Schemas
# ============================================================================
class TeamMemberBase(BaseModel):
"""Base team member schema."""
email: EmailStr = Field(..., description="Team member email address")
first_name: str | None = Field(None, max_length=100)
last_name: str | None = Field(None, max_length=100)
class TeamMemberInvite(TeamMemberBase):
"""Schema for inviting a team member."""
role_id: int | None = Field(
None, description="Role ID to assign (for preset roles)"
)
role_name: str | None = Field(
None, description="Role name (manager, staff, support, etc.)"
)
custom_permissions: list[str] | None = Field(
None, description="Custom permissions (overrides role preset)"
)
@field_validator("role_name")
def validate_role_name(cls, v):
"""Validate role name is in allowed presets."""
if v is not None:
allowed_roles = ["manager", "staff", "support", "viewer", "marketing"]
if v.lower() not in allowed_roles:
raise ValueError(
f"Role name must be one of: {', '.join(allowed_roles)}"
)
return v.lower() if v else v
@field_validator("custom_permissions")
def validate_custom_permissions(cls, v, values):
"""Ensure either role_id/role_name OR custom_permissions is provided."""
if v is not None and len(v) > 0:
# If custom permissions provided, role_name should be provided too
if "role_name" not in values or not values["role_name"]:
raise ValueError(
"role_name is required when providing custom_permissions"
)
return v
class TeamMemberUpdate(BaseModel):
"""Schema for updating a team member."""
role_id: int | None = Field(None, description="New role ID")
is_active: bool | None = Field(None, description="Active status")
class TeamMemberResponse(BaseModel):
"""Schema for team member response."""
id: int = Field(..., description="User ID")
email: EmailStr
username: str
first_name: str | None
last_name: str | None
full_name: str
user_type: str = Field(..., description="'owner' or 'member'")
role_name: str = Field(..., description="Role name")
role_id: int | None
permissions: list[str] = Field(
default_factory=list, description="User's permissions"
)
is_active: bool
is_owner: bool
invitation_pending: bool = Field(
default=False, description="True if invitation not yet accepted"
)
invited_at: datetime | None = Field(None, description="When invitation was sent")
accepted_at: datetime | None = Field(
None, description="When invitation was accepted"
)
joined_at: datetime = Field(..., description="When user joined vendor")
class Config:
from_attributes = True
class TeamMemberListResponse(BaseModel):
"""Schema for team member list response."""
members: list[TeamMemberResponse]
total: int
active_count: int
pending_invitations: int
# ============================================================================
# Invitation Schemas
# ============================================================================
class InvitationAccept(BaseModel):
"""Schema for accepting a team invitation."""
invitation_token: str = Field(
..., min_length=32, description="Invitation token from email"
)
password: str = Field(
..., min_length=8, max_length=128, description="Password for new account"
)
first_name: str = Field(..., min_length=1, max_length=100)
last_name: str = Field(..., min_length=1, max_length=100)
@field_validator("password")
def validate_password_strength(cls, v):
"""Validate password meets minimum requirements."""
if len(v) < 8:
raise ValueError("Password must be at least 8 characters long")
has_upper = any(c.isupper() for c in v)
has_lower = any(c.islower() for c in v)
has_digit = any(c.isdigit() for c in v)
if not (has_upper and has_lower and has_digit):
raise ValueError(
"Password must contain at least one uppercase letter, "
"one lowercase letter, and one digit"
)
return v
class InvitationResponse(BaseModel):
"""Schema for invitation response."""
message: str
email: EmailStr
role: str
invitation_token: str | None = Field(
None, description="Token (only returned in dev/test environments)"
)
invitation_sent: bool = Field(default=True)
class InvitationAcceptResponse(BaseModel):
"""Schema for invitation acceptance response."""
message: str
vendor: dict = Field(..., description="Vendor information")
user: dict = Field(..., description="User information")
role: str
# ============================================================================
# Team Statistics Schema
# ============================================================================
class TeamStatistics(BaseModel):
"""Schema for team statistics."""
total_members: int
active_members: int
inactive_members: int
pending_invitations: int
owners: int
team_members: int
roles_breakdown: dict = Field(
default_factory=dict, description="Count of members per role"
)
# ============================================================================
# Bulk Operations Schemas
# ============================================================================
class BulkRemoveRequest(BaseModel):
"""Schema for bulk removing team members."""
user_ids: list[int] = Field(
..., min_items=1, description="List of user IDs to remove"
)
class BulkRemoveResponse(BaseModel):
"""Schema for bulk remove response."""
success_count: int
failed_count: int
errors: list[dict] = Field(default_factory=list)
# ============================================================================
# Permission Check Schemas
# ============================================================================
class PermissionCheckRequest(BaseModel):
"""Schema for checking permissions."""
permissions: list[str] = Field(..., min_items=1, description="Permissions to check")
class PermissionCheckResponse(BaseModel):
"""Schema for permission check response."""
has_all: bool = Field(..., description="True if user has all permissions")
has_any: bool = Field(..., description="True if user has any permission")
granted: list[str] = Field(default_factory=list, description="Permissions user has")
denied: list[str] = Field(
default_factory=list, description="Permissions user lacks"
)
class UserPermissionsResponse(BaseModel):
"""Schema for user's permissions response."""
permissions: list[str] = Field(default_factory=list)
permission_count: int
is_owner: bool
role_name: str | None = None
# ============================================================================
# Error Response Schema
# ============================================================================
class TeamErrorResponse(BaseModel):
"""Schema for team operation errors."""
error_code: str
message: str
details: dict | None = None

351
models/schema/vendor.py
View File

@@ -1,351 +0,0 @@
# models/schema/vendor.py
"""
Pydantic schemas for Vendor-related operations.
Schemas include:
- VendorCreate: For creating vendors under companies
- VendorUpdate: For updating vendor information (Admin only)
- VendorResponse: Standard vendor response
- VendorDetailResponse: Vendor response with company/owner details
- VendorCreateResponse: Response after vendor creation
- VendorListResponse: Paginated vendor list
- VendorSummary: Lightweight vendor info
Note: Ownership transfer is handled at the Company level.
See models/schema/company.py for CompanyTransferOwnership.
"""
import re
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
class VendorCreate(BaseModel):
"""
Schema for creating a new vendor (storefront/brand) under an existing company.
Contact info is inherited from the parent company by default.
Optionally, provide contact fields to override from the start.
"""
# Parent company
company_id: int = Field(..., description="ID of the parent company", gt=0)
# Basic Information
vendor_code: str = Field(
...,
description="Unique vendor identifier (e.g., TECHSTORE)",
min_length=2,
max_length=50,
)
subdomain: str = Field(
..., description="Unique subdomain for the vendor", min_length=2, max_length=100
)
name: str = Field(
...,
description="Display name of the vendor/brand",
min_length=2,
max_length=255,
)
description: str | None = Field(None, description="Vendor/brand description")
# Platform assignments (optional - vendor can be on multiple platforms)
platform_ids: list[int] | None = Field(
None, description="List of platform IDs to assign the vendor to"
)
# Marketplace URLs (brand-specific multi-language support)
letzshop_csv_url_fr: str | None = Field(None, description="French CSV URL")
letzshop_csv_url_en: str | None = Field(None, description="English CSV URL")
letzshop_csv_url_de: str | None = Field(None, description="German CSV URL")
# Contact Info (optional - if not provided, inherited from company)
contact_email: str | None = Field(
None, description="Override company contact email"
)
contact_phone: str | None = Field(
None, description="Override company contact phone"
)
website: str | None = Field(None, description="Override company website")
business_address: str | None = Field(
None, description="Override company business address"
)
tax_number: str | None = Field(None, description="Override company tax number")
# Language Settings
default_language: str | None = Field(
"fr", description="Default language for content (en, fr, de, lb)"
)
dashboard_language: str | None = Field(
"fr", description="Vendor dashboard UI language"
)
storefront_language: str | None = Field(
"fr", description="Default storefront language for customers"
)
storefront_languages: list[str] | None = Field(
default=["fr", "de", "en"], description="Enabled languages for storefront"
)
storefront_locale: str | None = Field(
None,
description="Locale for currency/number formatting (e.g., 'fr-LU', 'de-DE'). NULL = inherit from platform default",
max_length=10,
)
@field_validator("subdomain")
@classmethod
def validate_subdomain(cls, v):
"""Validate subdomain format: lowercase alphanumeric with hyphens."""
if v and not re.match(r"^[a-z0-9][a-z0-9-]*[a-z0-9]$", v):
raise ValueError(
"Subdomain must contain only lowercase letters, numbers, and hyphens"
)
return v.lower() if v else v
@field_validator("vendor_code")
@classmethod
def validate_vendor_code(cls, v):
"""Ensure vendor code is uppercase for consistency."""
return v.upper() if v else v
class VendorUpdate(BaseModel):
"""
Schema for updating vendor information (Admin only).
Contact fields can be overridden at the vendor level.
Set to null/empty to reset to company default (inherit).
"""
# Basic Information
name: str | None = Field(None, min_length=2, max_length=255)
description: str | None = None
subdomain: str | None = Field(None, min_length=2, max_length=100)
# Marketplace URLs (brand-specific)
letzshop_csv_url_fr: str | None = None
letzshop_csv_url_en: str | None = None
letzshop_csv_url_de: str | None = None
# Status (Admin only)
is_active: bool | None = None
is_verified: bool | None = None
# Contact Info (set value to override, set to empty string to reset to inherit)
contact_email: str | None = Field(
None, description="Override company contact email"
)
contact_phone: str | None = Field(
None, description="Override company contact phone"
)
website: str | None = Field(None, description="Override company website")
business_address: str | None = Field(
None, description="Override company business address"
)
tax_number: str | None = Field(None, description="Override company tax number")
# Special flag to reset contact fields to inherit from company
reset_contact_to_company: bool | None = Field(
None, description="If true, reset all contact fields to inherit from company"
)
# Language Settings
default_language: str | None = Field(
None, description="Default language for content (en, fr, de, lb)"
)
dashboard_language: str | None = Field(
None, description="Vendor dashboard UI language"
)
storefront_language: str | None = Field(
None, description="Default storefront language for customers"
)
storefront_languages: list[str] | None = Field(
None, description="Enabled languages for storefront"
)
storefront_locale: str | None = Field(
None,
description="Locale for currency/number formatting (e.g., 'fr-LU', 'de-DE'). NULL = inherit from platform default",
max_length=10,
)
@field_validator("subdomain")
@classmethod
def subdomain_lowercase(cls, v):
"""Normalize subdomain to lowercase."""
return v.lower().strip() if v else v
model_config = ConfigDict(from_attributes=True)
class VendorResponse(BaseModel):
"""
Standard schema for vendor response data.
Note: Business contact info (contact_email, contact_phone, website,
business_address, tax_number) is now at the Company level.
Use company_id to look up company details.
"""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_code: str
subdomain: str
name: str
description: str | None
# Company relationship
company_id: int
# Marketplace URLs (brand-specific)
letzshop_csv_url_fr: str | None
letzshop_csv_url_en: str | None
letzshop_csv_url_de: str | None
# Status Flags
is_active: bool
is_verified: bool
# Language Settings (optional with defaults for backward compatibility)
default_language: str = "fr"
dashboard_language: str = "fr"
storefront_language: str = "fr"
storefront_languages: list[str] = ["fr", "de", "en"]
# Currency/number formatting locale (NULL = inherit from platform default)
storefront_locale: str | None = None
# Timestamps
created_at: datetime
updated_at: datetime
class VendorDetailResponse(VendorResponse):
"""
Extended vendor response including company information and resolved contact info.
Contact fields show the effective value (vendor override or company default)
with flags indicating if the value is inherited from the parent company.
"""
# Company info
company_name: str = Field(..., description="Name of the parent company")
# Owner info (at company level)
owner_email: str = Field(
..., description="Email of the company owner (for login/authentication)"
)
owner_username: str = Field(..., description="Username of the company owner")
# Resolved contact info (vendor override or company default)
contact_email: str | None = Field(None, description="Effective contact email")
contact_phone: str | None = Field(None, description="Effective contact phone")
website: str | None = Field(None, description="Effective website")
business_address: str | None = Field(None, description="Effective business address")
tax_number: str | None = Field(None, description="Effective tax number")
# Inheritance flags (True = value is inherited from company, not overridden)
contact_email_inherited: bool = Field(
False, description="True if contact_email is from company"
)
contact_phone_inherited: bool = Field(
False, description="True if contact_phone is from company"
)
website_inherited: bool = Field(
False, description="True if website is from company"
)
business_address_inherited: bool = Field(
False, description="True if business_address is from company"
)
tax_number_inherited: bool = Field(
False, description="True if tax_number is from company"
)
# Original company values (for reference in UI)
company_contact_email: str | None = Field(
None, description="Company's contact email"
)
company_contact_phone: str | None = Field(
None, description="Company's phone number"
)
company_website: str | None = Field(None, description="Company's website URL")
company_business_address: str | None = Field(
None, description="Company's business address"
)
company_tax_number: str | None = Field(None, description="Company's tax number")
class VendorCreateResponse(VendorDetailResponse):
"""
Response after creating vendor under an existing company.
The vendor is created under a company, so no new owner credentials are generated.
The company owner already has access to this vendor.
"""
login_url: str | None = Field(None, description="URL for vendor storefront")
class VendorListResponse(BaseModel):
"""Schema for paginated vendor list."""
vendors: list[VendorResponse]
total: int
skip: int
limit: int
class VendorSummary(BaseModel):
"""Lightweight vendor summary for dropdowns and quick references."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_code: str
subdomain: str
name: str
company_id: int
is_active: bool
# NOTE: Vendor ownership transfer schemas have been removed.
# Ownership transfer is now handled at the Company level.
# See models/schema/company.py for CompanyTransferOwnership and CompanyTransferOwnershipResponse.
# ============================================================================
# LETZSHOP EXPORT SCHEMAS
# ============================================================================
class LetzshopExportRequest(BaseModel):
"""Request body for Letzshop export to pickup folder."""
include_inactive: bool = Field(
default=False,
description="Include inactive products in export"
)
class LetzshopExportFileInfo(BaseModel):
"""Info about an exported file."""
language: str
filename: str | None = None
path: str | None = None
size_bytes: int | None = None
error: str | None = None
class LetzshopExportResponse(BaseModel):
"""Response from Letzshop export to folder."""
success: bool
message: str
vendor_code: str
export_directory: str
files: list[LetzshopExportFileInfo]
celery_task_id: str | None = None # Set when using Celery async export
is_async: bool = Field(default=False, serialization_alias="async") # True when queued via Celery
model_config = {"populate_by_name": True}

129
models/schema/vendor_domain.py
View File

@@ -1,129 +0,0 @@
# models/schema/vendor_domain.py
"""
Pydantic schemas for Vendor Domain operations.
Schemas include:
- VendorDomainCreate: For adding custom domains
- VendorDomainUpdate: For updating domain settings
- VendorDomainResponse: Standard domain response
- VendorDomainListResponse: Paginated domain list
- DomainVerificationInstructions: DNS verification instructions
"""
import re
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
class VendorDomainCreate(BaseModel):
"""Schema for adding a custom domain to vendor."""
domain: str = Field(
...,
description="Custom domain (e.g., myshop.com or shop.mybrand.com)",
min_length=3,
max_length=255,
)
is_primary: bool = Field(
default=False, description="Set as primary domain for the vendor"
)
@field_validator("domain")
@classmethod
def validate_domain(cls, v: str) -> str:
"""Validate and normalize domain."""
# Remove protocol if present
domain = v.replace("https://", "").replace("http://", "") # noqa: SEC-034
# Remove trailing slash
domain = domain.rstrip("/")
# Convert to lowercase
domain = domain.lower().strip()
# Basic validation
if not domain or "/" in domain:
raise ValueError("Invalid domain format")
if "." not in domain:
raise ValueError("Domain must have at least one dot")
# Check for reserved subdomains
reserved = ["www", "admin", "api", "mail", "smtp", "ftp", "cpanel", "webmail"]
first_part = domain.split(".")[0]
if first_part in reserved:
raise ValueError(
f"Domain cannot start with reserved subdomain: {first_part}"
)
# Validate domain format (basic regex)
domain_pattern = r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$"
if not re.match(domain_pattern, domain):
raise ValueError("Invalid domain format")
return domain
class VendorDomainUpdate(BaseModel):
"""Schema for updating vendor domain settings."""
is_primary: bool | None = Field(None, description="Set as primary domain")
is_active: bool | None = Field(None, description="Activate or deactivate domain")
model_config = ConfigDict(from_attributes=True)
class VendorDomainResponse(BaseModel):
"""Standard schema for vendor domain response."""
model_config = ConfigDict(from_attributes=True)
id: int
vendor_id: int
domain: str
is_primary: bool
is_active: bool
is_verified: bool
ssl_status: str
verification_token: str | None = None
verified_at: datetime | None = None
ssl_verified_at: datetime | None = None
created_at: datetime
updated_at: datetime
class VendorDomainListResponse(BaseModel):
"""Schema for paginated vendor domain list."""
domains: list[VendorDomainResponse]
total: int
class DomainVerificationInstructions(BaseModel):
"""DNS verification instructions for domain ownership."""
domain: str
verification_token: str
instructions: dict[str, str]
txt_record: dict[str, str]
common_registrars: dict[str, str]
model_config = ConfigDict(from_attributes=True)
class DomainVerificationResponse(BaseModel):
"""Response after domain verification."""
message: str
domain: str
verified_at: datetime
is_verified: bool
class DomainDeletionResponse(BaseModel):
"""Response after domain deletion."""
message: str
domain: str
vendor_id: int

108
models/schema/vendor_theme.py
View File

@@ -1,108 +0,0 @@
# models/schema/vendor_theme.py
"""
Pydantic schemas for vendor theme operations.
"""
from pydantic import BaseModel, Field
class VendorThemeColors(BaseModel):
"""Color scheme for vendor theme."""
primary: str | None = Field(None, description="Primary brand color")
secondary: str | None = Field(None, description="Secondary color")
accent: str | None = Field(None, description="Accent/CTA color")
background: str | None = Field(None, description="Background color")
text: str | None = Field(None, description="Text color")
border: str | None = Field(None, description="Border color")
class VendorThemeFonts(BaseModel):
"""Typography settings for vendor theme."""
heading: str | None = Field(None, description="Font for headings")
body: str | None = Field(None, description="Font for body text")
class VendorThemeBranding(BaseModel):
"""Branding assets for vendor theme."""
logo: str | None = Field(None, description="Logo URL")
logo_dark: str | None = Field(None, description="Dark mode logo URL")
favicon: str | None = Field(None, description="Favicon URL")
banner: str | None = Field(None, description="Banner image URL")
class VendorThemeLayout(BaseModel):
"""Layout settings for vendor theme."""
style: str | None = Field(
None, description="Product layout style (grid, list, masonry)"
)
header: str | None = Field(
None, description="Header style (fixed, static, transparent)"
)
product_card: str | None = Field(
None, description="Product card style (modern, classic, minimal)"
)
class VendorThemeUpdate(BaseModel):
"""Schema for updating vendor theme (partial updates allowed)."""
theme_name: str | None = Field(None, description="Theme preset name")
colors: dict[str, str] | None = Field(None, description="Color scheme")
fonts: dict[str, str] | None = Field(None, description="Font settings")
branding: dict[str, str | None] | None = Field(None, description="Branding assets")
layout: dict[str, str] | None = Field(None, description="Layout settings")
custom_css: str | None = Field(None, description="Custom CSS rules")
social_links: dict[str, str] | None = Field(None, description="Social media links")
class VendorThemeResponse(BaseModel):
"""Schema for vendor theme response."""
theme_name: str = Field(..., description="Theme name")
colors: dict[str, str] = Field(..., description="Color scheme")
fonts: dict[str, str] = Field(..., description="Font settings")
branding: dict[str, str | None] = Field(..., description="Branding assets")
layout: dict[str, str] = Field(..., description="Layout settings")
social_links: dict[str, str] | None = Field(
default_factory=dict, description="Social links"
)
custom_css: str | None = Field(None, description="Custom CSS")
css_variables: dict[str, str] | None = Field(
None, description="CSS custom properties"
)
class ThemePresetPreview(BaseModel):
"""Preview information for a theme preset."""
name: str = Field(..., description="Preset name")
description: str = Field(..., description="Preset description")
primary_color: str = Field(..., description="Primary color")
secondary_color: str = Field(..., description="Secondary color")
accent_color: str = Field(..., description="Accent color")
heading_font: str = Field(..., description="Heading font")
body_font: str = Field(..., description="Body font")
layout_style: str = Field(..., description="Layout style")
class ThemePresetResponse(BaseModel):
"""Response after applying a preset."""
message: str = Field(..., description="Success message")
theme: VendorThemeResponse = Field(..., description="Applied theme")
class ThemePresetListResponse(BaseModel):
"""List of available theme presets."""
presets: list[ThemePresetPreview] = Field(..., description="Available presets")
class ThemeDeleteResponse(BaseModel):
"""Response after deleting a theme."""
message: str = Field(..., description="Success message")