- Enhanced route discovery system with ROUTE_CONFIG support for custom
prefix, tags, and priority
- Added get_admin_api_routes() and get_vendor_api_routes() helpers that
return routes sorted by priority
- Added fallback discovery for routes/{frontend}.py when routes/api/
doesn't exist
- Updated CMS module with ROUTE_CONFIG (prefix: /content-pages,
priority: 100) to register last for catch-all routes
- Moved customers routes from routes/ to routes/api/ directory
- Updated orders module to aggregate exception routers into main routers
- Removed manual module router imports from admin and vendor API init
files, replaced with auto-discovery loop
Modules now auto-discovered: billing, inventory, orders, marketplace,
cms, customers, analytics, loyalty, messaging, monitoring, dev-tools
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
317 lines
9.8 KiB
Python
317 lines
9.8 KiB
Python
# app/modules/routes.py
|
|
"""
|
|
Module route discovery for FastAPI.
|
|
|
|
Provides utilities to:
|
|
- Discover route modules from all registered modules
|
|
- Auto-register API and page routes from self-contained modules
|
|
|
|
This module bridges the gap between the module system and FastAPI,
|
|
allowing routes to be defined within modules and automatically
|
|
discovered and registered.
|
|
|
|
Usage:
|
|
# In main.py
|
|
from app.modules.routes import discover_module_routes
|
|
|
|
# Auto-discover and register routes
|
|
for route_info in discover_module_routes():
|
|
app.include_router(
|
|
route_info["router"],
|
|
prefix=route_info["prefix"],
|
|
tags=route_info["tags"],
|
|
include_in_schema=route_info.get("include_in_schema", True),
|
|
)
|
|
|
|
Route Configuration:
|
|
Modules can export a ROUTE_CONFIG dict to customize route registration:
|
|
|
|
ROUTE_CONFIG = {
|
|
"prefix": "/content-pages", # Custom prefix (replaces default)
|
|
"tags": ["admin-content-pages"], # Custom tags
|
|
"priority": 100, # Higher = registered later (for catch-all routes)
|
|
}
|
|
"""
|
|
|
|
import importlib
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import APIRouter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class RouteInfo:
|
|
"""Information about a discovered route."""
|
|
|
|
router: "APIRouter"
|
|
prefix: str
|
|
tags: list[str]
|
|
include_in_schema: bool = True
|
|
module_code: str = ""
|
|
route_type: str = "" # "api" or "pages"
|
|
frontend: str = "" # "admin", "vendor", "shop"
|
|
priority: int = 0 # Higher = registered later (for catch-all routes)
|
|
custom_prefix: str = "" # Custom prefix from ROUTE_CONFIG
|
|
|
|
|
|
def discover_module_routes() -> list[RouteInfo]:
|
|
"""
|
|
Discover routes from all registered self-contained modules.
|
|
|
|
Scans all modules in the registry and returns RouteInfo objects for
|
|
modules that have routes directories.
|
|
|
|
Route discovery looks for:
|
|
- routes/api/admin.py -> admin API routes
|
|
- routes/api/vendor.py -> vendor API routes
|
|
- routes/api/shop.py -> shop API routes
|
|
- routes/pages/admin.py -> admin page routes
|
|
- routes/pages/vendor.py -> vendor page routes
|
|
|
|
Returns:
|
|
List of RouteInfo objects with router and registration info
|
|
|
|
Example:
|
|
>>> routes = discover_module_routes()
|
|
>>> for route in routes:
|
|
... print(f"{route.module_code}: {route.route_type}/{route.frontend}")
|
|
analytics: pages/vendor
|
|
cms: api/admin
|
|
cms: api/vendor
|
|
cms: pages/admin
|
|
cms: pages/vendor
|
|
"""
|
|
# Import here to avoid circular imports
|
|
from app.modules.registry import MODULES
|
|
|
|
routes: list[RouteInfo] = []
|
|
|
|
for module in MODULES.values():
|
|
if not module.is_self_contained:
|
|
continue
|
|
|
|
module_routes = _discover_module_routes(module.code)
|
|
routes.extend(module_routes)
|
|
|
|
logger.info(f"Discovered {len(routes)} routes from self-contained modules")
|
|
return routes
|
|
|
|
|
|
def _discover_module_routes(module_code: str) -> list[RouteInfo]:
|
|
"""
|
|
Discover routes for a specific module.
|
|
|
|
Args:
|
|
module_code: Module code (e.g., "analytics", "cms")
|
|
|
|
Returns:
|
|
List of RouteInfo for this module
|
|
"""
|
|
routes: list[RouteInfo] = []
|
|
dir_name = module_code.replace("-", "_")
|
|
module_path = Path(__file__).parent / dir_name
|
|
|
|
if not module_path.exists():
|
|
return routes
|
|
|
|
routes_path = module_path / "routes"
|
|
if not routes_path.exists():
|
|
return routes
|
|
|
|
# Discover API routes (with fallback to routes/ for legacy modules)
|
|
api_routes = _discover_routes_in_dir(
|
|
module_code, dir_name, routes_path / "api", "api",
|
|
fallback_dir=routes_path # Allow routes/admin.py as fallback
|
|
)
|
|
routes.extend(api_routes)
|
|
|
|
# Discover page routes
|
|
page_routes = _discover_routes_in_dir(
|
|
module_code, dir_name, routes_path / "pages", "pages"
|
|
)
|
|
routes.extend(page_routes)
|
|
|
|
return routes
|
|
|
|
|
|
def _discover_routes_in_dir(
|
|
module_code: str, dir_name: str, routes_dir: Path, route_type: str,
|
|
fallback_dir: Path | None = None
|
|
) -> list[RouteInfo]:
|
|
"""
|
|
Discover routes in a specific directory (api/ or pages/).
|
|
|
|
Args:
|
|
module_code: Module code for tags
|
|
dir_name: Directory name (module_code with _ instead of -)
|
|
routes_dir: Path to routes/api/ or routes/pages/
|
|
route_type: "api" or "pages"
|
|
fallback_dir: Optional fallback directory (e.g., routes/ for modules
|
|
with routes/admin.py instead of routes/api/admin.py)
|
|
|
|
Returns:
|
|
List of RouteInfo for discovered routes
|
|
"""
|
|
routes: list[RouteInfo] = []
|
|
|
|
# Look for admin.py, vendor.py, shop.py
|
|
frontends = {
|
|
"admin": {
|
|
"api_prefix": "/api/v1/admin",
|
|
"pages_prefix": "/admin",
|
|
"include_in_schema": True,
|
|
},
|
|
"vendor": {
|
|
"api_prefix": "/api/v1/vendor",
|
|
"pages_prefix": "/vendor",
|
|
"include_in_schema": True if route_type == "api" else False,
|
|
},
|
|
"shop": {
|
|
"api_prefix": "/api/v1/shop",
|
|
"pages_prefix": "/shop",
|
|
"include_in_schema": True if route_type == "api" else False,
|
|
},
|
|
}
|
|
|
|
for frontend, config in frontends.items():
|
|
# Check primary location first, then fallback
|
|
route_file = routes_dir / f"{frontend}.py" if routes_dir.exists() else None
|
|
use_fallback = False
|
|
|
|
if route_file is None or not route_file.exists():
|
|
if fallback_dir and fallback_dir.exists():
|
|
fallback_file = fallback_dir / f"{frontend}.py"
|
|
if fallback_file.exists():
|
|
route_file = fallback_file
|
|
use_fallback = True
|
|
|
|
if route_file is None or not route_file.exists():
|
|
continue
|
|
|
|
try:
|
|
# Import the module
|
|
if use_fallback:
|
|
import_path = f"app.modules.{dir_name}.routes.{frontend}"
|
|
else:
|
|
import_path = f"app.modules.{dir_name}.routes.{route_type}.{frontend}"
|
|
route_module = importlib.import_module(import_path)
|
|
|
|
# Get the router (try common names)
|
|
router = None
|
|
for attr_name in ["router", f"{frontend}_router", "api_router", "page_router"]:
|
|
if hasattr(route_module, attr_name):
|
|
router = getattr(route_module, attr_name)
|
|
break
|
|
|
|
if router is None:
|
|
logger.warning(f"No router found in {import_path}")
|
|
continue
|
|
|
|
# Read ROUTE_CONFIG if present
|
|
route_config = getattr(route_module, "ROUTE_CONFIG", {})
|
|
custom_prefix = route_config.get("prefix", "")
|
|
custom_tags = route_config.get("tags", [])
|
|
priority = route_config.get("priority", 0)
|
|
|
|
# Determine prefix based on route type
|
|
if route_type == "api":
|
|
prefix = config["api_prefix"]
|
|
else:
|
|
prefix = config["pages_prefix"]
|
|
|
|
# Build tags - use custom tags if provided, otherwise default
|
|
if custom_tags:
|
|
tags = custom_tags
|
|
else:
|
|
tags = [f"{frontend}-{module_code}"]
|
|
|
|
route_info = RouteInfo(
|
|
router=router,
|
|
prefix=prefix,
|
|
tags=tags,
|
|
include_in_schema=config["include_in_schema"],
|
|
module_code=module_code,
|
|
route_type=route_type,
|
|
frontend=frontend,
|
|
priority=priority,
|
|
custom_prefix=custom_prefix,
|
|
)
|
|
routes.append(route_info)
|
|
|
|
logger.debug(f"Discovered route: {module_code} {route_type}/{frontend}")
|
|
|
|
except ImportError as e:
|
|
logger.warning(f"Failed to import {import_path}: {e}")
|
|
except Exception as e:
|
|
logger.error(f"Error discovering routes in {route_file}: {e}")
|
|
|
|
return routes
|
|
|
|
|
|
def get_api_routes() -> list[RouteInfo]:
|
|
"""Get only API routes from modules."""
|
|
return [r for r in discover_module_routes() if r.route_type == "api"]
|
|
|
|
|
|
def get_page_routes() -> list[RouteInfo]:
|
|
"""Get only page routes from modules."""
|
|
return [r for r in discover_module_routes() if r.route_type == "pages"]
|
|
|
|
|
|
def get_vendor_page_routes() -> list[RouteInfo]:
|
|
"""Get vendor page routes from modules."""
|
|
return [r for r in discover_module_routes() if r.route_type == "pages" and r.frontend == "vendor"]
|
|
|
|
|
|
def get_admin_page_routes() -> list[RouteInfo]:
|
|
"""Get admin page routes from modules."""
|
|
return [r for r in discover_module_routes() if r.route_type == "pages" and r.frontend == "admin"]
|
|
|
|
|
|
def get_admin_api_routes() -> list[RouteInfo]:
|
|
"""
|
|
Get admin API routes from modules, sorted by priority.
|
|
|
|
Returns routes sorted by priority (lower first, higher last).
|
|
This ensures catch-all routes (priority 100+) are registered after
|
|
specific routes.
|
|
"""
|
|
routes = [
|
|
r for r in discover_module_routes()
|
|
if r.route_type == "api" and r.frontend == "admin"
|
|
]
|
|
return sorted(routes, key=lambda r: r.priority)
|
|
|
|
|
|
def get_vendor_api_routes() -> list[RouteInfo]:
|
|
"""
|
|
Get vendor API routes from modules, sorted by priority.
|
|
|
|
Returns routes sorted by priority (lower first, higher last).
|
|
This ensures catch-all routes (priority 100+) are registered after
|
|
specific routes.
|
|
"""
|
|
routes = [
|
|
r for r in discover_module_routes()
|
|
if r.route_type == "api" and r.frontend == "vendor"
|
|
]
|
|
return sorted(routes, key=lambda r: r.priority)
|
|
|
|
|
|
__all__ = [
|
|
"RouteInfo",
|
|
"discover_module_routes",
|
|
"get_api_routes",
|
|
"get_page_routes",
|
|
"get_vendor_page_routes",
|
|
"get_admin_page_routes",
|
|
"get_admin_api_routes",
|
|
"get_vendor_api_routes",
|
|
]
|