feat: add module config and migrations auto-discovery infrastructure
Add self-contained configuration and migrations support for modules:
Config auto-discovery (app/modules/config.py):
- Modules can have config.py with Pydantic Settings
- Environment variables prefixed with MODULE_NAME_
- Auto-discovered via get_module_config()
Migrations auto-discovery:
- Each module has migrations/versions/ directory
- Alembic discovers module migrations automatically
- Naming convention: {module}_{seq}_{description}.py
New architecture rules (MOD-013 to MOD-015):
- MOD-013: config.py should export config/config_class
- MOD-014: Migrations must follow naming convention
- MOD-015: Migrations directory must have __init__.py
Created for all 11 self-contained modules:
- config.py placeholder files
- migrations/ directories with __init__.py files
Added core and tenancy module definitions for completeness.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
248
app/modules/routes.py
Normal file
248
app/modules/routes.py
Normal file
@@ -0,0 +1,248 @@
|
||||
# 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),
|
||||
)
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
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"
|
||||
|
||||
|
||||
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
|
||||
api_routes = _discover_routes_in_dir(
|
||||
module_code, dir_name, routes_path / "api", "api"
|
||||
)
|
||||
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
|
||||
) -> 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"
|
||||
|
||||
Returns:
|
||||
List of RouteInfo for discovered routes
|
||||
"""
|
||||
routes: list[RouteInfo] = []
|
||||
|
||||
if not routes_dir.exists():
|
||||
return routes
|
||||
|
||||
# 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():
|
||||
route_file = routes_dir / f"{frontend}.py"
|
||||
if not route_file.exists():
|
||||
continue
|
||||
|
||||
try:
|
||||
# Import the module
|
||||
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
|
||||
|
||||
# Determine prefix based on route type
|
||||
if route_type == "api":
|
||||
prefix = config["api_prefix"]
|
||||
else:
|
||||
prefix = config["pages_prefix"]
|
||||
|
||||
# Build tags
|
||||
tags = [f"{module_code}-{frontend}-{route_type}"]
|
||||
|
||||
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,
|
||||
)
|
||||
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"]
|
||||
|
||||
|
||||
__all__ = [
|
||||
"RouteInfo",
|
||||
"discover_module_routes",
|
||||
"get_api_routes",
|
||||
"get_page_routes",
|
||||
"get_vendor_page_routes",
|
||||
"get_admin_page_routes",
|
||||
]
|
||||
Reference in New Issue
Block a user