refactor(arch): eliminate all cross-module model imports in service layer
Some checks failed
CI / ruff (push) Successful in 9s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled
CI / validate (push) Has been cancelled
CI / dependency-scanning (push) Has been cancelled
CI / pytest (push) Has been cancelled

Enforce MOD-025/MOD-026 rules: zero top-level cross-module model imports
remain in any service file. All 66 files migrated using deferred import
patterns (method-body, _get_model() helpers, instance-cached self._Model)
and new cross-module service methods in tenancy. Documentation updated
with Pattern 6 (deferred imports), migration plan marked complete, and
violations status reflects 84→0 service-layer violations.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-27 06:13:15 +01:00
parent e3a52f6536
commit 86e85a98b8
66 changed files with 2242 additions and 1295 deletions

View File

@@ -17,7 +17,6 @@ from dataclasses import dataclass
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.modules.cms.models import ContentPage
from app.modules.tenancy.exceptions import (
PlatformNotFoundException,
)
@@ -102,6 +101,11 @@ class PlatformService:
return platform
@staticmethod
def get_default_platform(db: Session) -> Platform | None:
"""Get the first/default platform."""
return db.query(Platform).first()
@staticmethod
def list_platforms(
db: Session, include_inactive: bool = False
@@ -167,6 +171,13 @@ class PlatformService:
or 0
)
@staticmethod
def _get_content_page_model():
"""Deferred import for CMS ContentPage model."""
from app.modules.cms.models import ContentPage
return ContentPage
@staticmethod
def get_platform_pages_count(db: Session, platform_id: int) -> int:
"""
@@ -179,6 +190,7 @@ class PlatformService:
Returns:
Platform pages count
"""
ContentPage = PlatformService._get_content_page_model()
return (
db.query(func.count(ContentPage.id))
.filter(
@@ -202,6 +214,7 @@ class PlatformService:
Returns:
Store defaults count
"""
ContentPage = PlatformService._get_content_page_model()
return (
db.query(func.count(ContentPage.id))
.filter(
@@ -225,6 +238,7 @@ class PlatformService:
Returns:
Store overrides count
"""
ContentPage = PlatformService._get_content_page_model()
return (
db.query(func.count(ContentPage.id))
.filter(
@@ -247,6 +261,7 @@ class PlatformService:
Returns:
Published pages count
"""
ContentPage = PlatformService._get_content_page_model()
return (
db.query(func.count(ContentPage.id))
.filter(
@@ -269,6 +284,7 @@ class PlatformService:
Returns:
Draft pages count
"""
ContentPage = PlatformService._get_content_page_model()
return (
db.query(func.count(ContentPage.id))
.filter(
@@ -303,6 +319,187 @@ class PlatformService:
draft_pages_count=cls.get_draft_pages_count(db, platform.id),
)
# ========================================================================
# StorePlatform cross-module public API methods
# ========================================================================
@staticmethod
def get_primary_platform_id_for_store(db: Session, store_id: int) -> int | None:
"""
Get the primary platform ID for a store.
Args:
db: Database session
store_id: Store ID
Returns:
Platform ID or None if no platform assigned
"""
result = (
db.query(StorePlatform.platform_id)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.is_active == True, # noqa: E712
)
.order_by(StorePlatform.is_primary.desc())
.first()
)
return result[0] if result else None
@staticmethod
def get_active_platform_ids_for_store(db: Session, store_id: int) -> list[int]:
"""
Get all active platform IDs for a store.
Args:
db: Database session
store_id: Store ID
Returns:
List of platform IDs
"""
results = (
db.query(StorePlatform.platform_id)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.is_active == True, # noqa: E712
)
.order_by(StorePlatform.is_primary.desc())
.all()
)
return [r[0] for r in results]
@staticmethod
def get_store_platform_entry(
db: Session, store_id: int, platform_id: int
) -> StorePlatform | None:
"""
Get a specific StorePlatform entry.
Args:
db: Database session
store_id: Store ID
platform_id: Platform ID
Returns:
StorePlatform object or None
"""
return (
db.query(StorePlatform)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.platform_id == platform_id,
)
.first()
)
@staticmethod
def get_primary_store_platform_entry(
db: Session, store_id: int
) -> StorePlatform | None:
"""
Get the primary StorePlatform entry for a store.
Args:
db: Database session
store_id: Store ID
Returns:
StorePlatform object or None
"""
return (
db.query(StorePlatform)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.is_primary.is_(True),
)
.first()
)
@staticmethod
def get_store_ids_for_platform(
db: Session, platform_id: int, active_only: bool = True
) -> list[int]:
"""
Get store IDs subscribed to a platform.
Args:
db: Database session
platform_id: Platform ID
active_only: Only return active store-platform links
Returns:
List of store IDs
"""
query = db.query(StorePlatform.store_id).filter(
StorePlatform.platform_id == platform_id,
)
if active_only:
query = query.filter(StorePlatform.is_active == True) # noqa: E712
return [r[0] for r in query.all()]
@staticmethod
def ensure_store_platform(
db: Session,
store_id: int,
platform_id: int,
is_active: bool,
tier_id: int | None = None,
) -> StorePlatform | None:
"""
Upsert a StorePlatform entry.
If the entry exists, update is_active (and tier_id if provided).
If missing and is_active=True, create it (set is_primary if store has none).
If missing and is_active=False, no-op.
Args:
db: Database session
store_id: Store ID
platform_id: Platform ID
is_active: Whether the store-platform link is active
tier_id: Optional subscription tier ID
Returns:
The StorePlatform entry, or None if no-op
"""
existing = (
db.query(StorePlatform)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.platform_id == platform_id,
)
.first()
)
if existing:
existing.is_active = is_active
if tier_id is not None:
existing.tier_id = tier_id
return existing
if is_active:
has_primary = (
db.query(StorePlatform)
.filter(
StorePlatform.store_id == store_id,
StorePlatform.is_primary.is_(True),
)
.first()
) is not None
sp = StorePlatform(
store_id=store_id,
platform_id=platform_id,
is_active=True,
is_primary=not has_primary,
tier_id=tier_id,
)
db.add(sp)
return sp
return None
@staticmethod
def update_platform(
db: Session, platform: Platform, update_data: dict