refactor: migrate remaining routes to modules and enforce auto-discovery
MIGRATION: - Delete app/api/v1/vendor/analytics.py (duplicate - analytics module already auto-discovered) - Move usage routes from app/api/v1/vendor/usage.py to billing module - Move onboarding routes from app/api/v1/vendor/onboarding.py to marketplace module - Move features routes to billing module (admin + vendor) - Move inventory routes to inventory module (admin + vendor) - Move marketplace/letzshop routes to marketplace module - Move orders routes to orders module - Delete legacy letzshop service files (moved to marketplace module) DOCUMENTATION: - Add docs/development/migration/module-autodiscovery-migration.md with full migration history - Update docs/architecture/module-system.md with Entity Auto-Discovery Reference section - Add detailed sections for each entity type: routes, services, models, schemas, tasks, exceptions, templates, static files, locales, configuration ARCHITECTURE VALIDATION: - Add MOD-016: Routes must be in modules, not app/api/v1/ - Add MOD-017: Services must be in modules, not app/services/ - Add MOD-018: Tasks must be in modules, not app/tasks/ - Add MOD-019: Schemas must be in modules, not models/schema/ - Update scripts/validate_architecture.py with _validate_legacy_locations method - Update .architecture-rules/module.yaml with legacy location rules These rules enforce that all entities must be in self-contained modules. Legacy locations now trigger ERROR severity violations. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -3,8 +3,10 @@
|
||||
Billing module API routes.
|
||||
|
||||
Provides REST API endpoints for subscription and billing management:
|
||||
- Admin API: Subscription tier management, vendor subscriptions, billing history
|
||||
- Vendor API: Subscription status, tier comparison, invoices
|
||||
- Admin API: Subscription tier management, vendor subscriptions, billing history, features
|
||||
- Vendor API: Subscription status, tier comparison, invoices, features
|
||||
|
||||
Each main router (admin.py, vendor.py) aggregates its related sub-routers internally.
|
||||
"""
|
||||
|
||||
from app.modules.billing.routes.api.admin import admin_router
|
||||
|
||||
@@ -334,3 +334,12 @@ def update_vendor_subscription(
|
||||
products_count=usage["products_count"],
|
||||
team_count=usage["team_count"],
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Aggregate Feature Management Routes
|
||||
# ============================================================================
|
||||
# Include the features router to aggregate all billing-related admin routes
|
||||
from app.modules.billing.routes.api.admin_features import admin_features_router
|
||||
|
||||
admin_router.include_router(admin_features_router, tags=["admin-features"])
|
||||
|
||||
313
app/modules/billing/routes/api/admin_features.py
Normal file
313
app/modules/billing/routes/api/admin_features.py
Normal file
@@ -0,0 +1,313 @@
|
||||
# app/modules/billing/routes/api/admin_features.py
|
||||
"""
|
||||
Admin feature management endpoints.
|
||||
|
||||
Provides endpoints for:
|
||||
- Listing all features with their tier assignments
|
||||
- Updating tier feature assignments
|
||||
- Managing feature metadata
|
||||
- Viewing feature usage statistics
|
||||
|
||||
All routes require module access control for the 'billing' module.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.deps import get_current_admin_api, require_module_access
|
||||
from app.core.database import get_db
|
||||
from app.services.feature_service import feature_service
|
||||
from models.schema.auth import UserContext
|
||||
|
||||
admin_features_router = APIRouter(
|
||||
prefix="/features",
|
||||
dependencies=[Depends(require_module_access("billing"))],
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Response Schemas
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class FeatureResponse(BaseModel):
|
||||
"""Feature information for admin."""
|
||||
|
||||
id: int
|
||||
code: str
|
||||
name: str
|
||||
description: str | None = None
|
||||
category: str
|
||||
ui_location: str | None = None
|
||||
ui_icon: str | None = None
|
||||
ui_route: str | None = None
|
||||
ui_badge_text: str | None = None
|
||||
minimum_tier_id: int | None = None
|
||||
minimum_tier_code: str | None = None
|
||||
minimum_tier_name: str | None = None
|
||||
is_active: bool
|
||||
is_visible: bool
|
||||
display_order: int
|
||||
|
||||
|
||||
class FeatureListResponse(BaseModel):
|
||||
"""List of features."""
|
||||
|
||||
features: list[FeatureResponse]
|
||||
total: int
|
||||
|
||||
|
||||
class TierFeaturesResponse(BaseModel):
|
||||
"""Tier with its features."""
|
||||
|
||||
id: int
|
||||
code: str
|
||||
name: str
|
||||
description: str | None = None
|
||||
features: list[str]
|
||||
feature_count: int
|
||||
|
||||
|
||||
class TierListWithFeaturesResponse(BaseModel):
|
||||
"""All tiers with their features."""
|
||||
|
||||
tiers: list[TierFeaturesResponse]
|
||||
|
||||
|
||||
class UpdateTierFeaturesRequest(BaseModel):
|
||||
"""Request to update tier features."""
|
||||
|
||||
feature_codes: list[str]
|
||||
|
||||
|
||||
class UpdateFeatureRequest(BaseModel):
|
||||
"""Request to update feature metadata."""
|
||||
|
||||
name: str | None = None
|
||||
description: str | None = None
|
||||
category: str | None = None
|
||||
ui_location: str | None = None
|
||||
ui_icon: str | None = None
|
||||
ui_route: str | None = None
|
||||
ui_badge_text: str | None = None
|
||||
minimum_tier_code: str | None = None
|
||||
is_active: bool | None = None
|
||||
is_visible: bool | None = None
|
||||
display_order: int | None = None
|
||||
|
||||
|
||||
class CategoryListResponse(BaseModel):
|
||||
"""List of feature categories."""
|
||||
|
||||
categories: list[str]
|
||||
|
||||
|
||||
class TierFeatureDetailResponse(BaseModel):
|
||||
"""Tier features with full details."""
|
||||
|
||||
tier_code: str
|
||||
tier_name: str
|
||||
features: list[dict]
|
||||
feature_count: int
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Helper Functions
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def _feature_to_response(feature) -> FeatureResponse:
|
||||
"""Convert Feature model to response."""
|
||||
return FeatureResponse(
|
||||
id=feature.id,
|
||||
code=feature.code,
|
||||
name=feature.name,
|
||||
description=feature.description,
|
||||
category=feature.category,
|
||||
ui_location=feature.ui_location,
|
||||
ui_icon=feature.ui_icon,
|
||||
ui_route=feature.ui_route,
|
||||
ui_badge_text=feature.ui_badge_text,
|
||||
minimum_tier_id=feature.minimum_tier_id,
|
||||
minimum_tier_code=feature.minimum_tier.code if feature.minimum_tier else None,
|
||||
minimum_tier_name=feature.minimum_tier.name if feature.minimum_tier else None,
|
||||
is_active=feature.is_active,
|
||||
is_visible=feature.is_visible,
|
||||
display_order=feature.display_order,
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Endpoints
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@admin_features_router.get("", response_model=FeatureListResponse)
|
||||
def list_features(
|
||||
category: str | None = Query(None, description="Filter by category"),
|
||||
active_only: bool = Query(False, description="Only active features"),
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all features with their tier assignments."""
|
||||
features = feature_service.get_all_features(
|
||||
db, category=category, active_only=active_only
|
||||
)
|
||||
|
||||
return FeatureListResponse(
|
||||
features=[_feature_to_response(f) for f in features],
|
||||
total=len(features),
|
||||
)
|
||||
|
||||
|
||||
@admin_features_router.get("/categories", response_model=CategoryListResponse)
|
||||
def list_categories(
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all feature categories."""
|
||||
categories = feature_service.get_categories(db)
|
||||
return CategoryListResponse(categories=categories)
|
||||
|
||||
|
||||
@admin_features_router.get("/tiers", response_model=TierListWithFeaturesResponse)
|
||||
def list_tiers_with_features(
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""List all tiers with their feature assignments."""
|
||||
tiers = feature_service.get_all_tiers_with_features(db)
|
||||
|
||||
return TierListWithFeaturesResponse(
|
||||
tiers=[
|
||||
TierFeaturesResponse(
|
||||
id=t.id,
|
||||
code=t.code,
|
||||
name=t.name,
|
||||
description=t.description,
|
||||
features=t.features or [],
|
||||
feature_count=len(t.features or []),
|
||||
)
|
||||
for t in tiers
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@admin_features_router.get("/{feature_code}", response_model=FeatureResponse)
|
||||
def get_feature(
|
||||
feature_code: str,
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get a single feature by code.
|
||||
|
||||
Raises 404 if feature not found.
|
||||
"""
|
||||
feature = feature_service.get_feature_by_code(db, feature_code)
|
||||
|
||||
if not feature:
|
||||
from app.exceptions import FeatureNotFoundError
|
||||
|
||||
raise FeatureNotFoundError(feature_code)
|
||||
|
||||
return _feature_to_response(feature)
|
||||
|
||||
|
||||
@admin_features_router.put("/{feature_code}", response_model=FeatureResponse)
|
||||
def update_feature(
|
||||
feature_code: str,
|
||||
request: UpdateFeatureRequest,
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Update feature metadata.
|
||||
|
||||
Raises 404 if feature not found, 400 if tier code is invalid.
|
||||
"""
|
||||
feature = feature_service.update_feature(
|
||||
db,
|
||||
feature_code,
|
||||
name=request.name,
|
||||
description=request.description,
|
||||
category=request.category,
|
||||
ui_location=request.ui_location,
|
||||
ui_icon=request.ui_icon,
|
||||
ui_route=request.ui_route,
|
||||
ui_badge_text=request.ui_badge_text,
|
||||
minimum_tier_code=request.minimum_tier_code,
|
||||
is_active=request.is_active,
|
||||
is_visible=request.is_visible,
|
||||
display_order=request.display_order,
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.refresh(feature)
|
||||
|
||||
logger.info(f"Updated feature {feature_code} by admin {current_user.id}")
|
||||
|
||||
return _feature_to_response(feature)
|
||||
|
||||
|
||||
@admin_features_router.put("/tiers/{tier_code}/features", response_model=TierFeaturesResponse)
|
||||
def update_tier_features(
|
||||
tier_code: str,
|
||||
request: UpdateTierFeaturesRequest,
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Update features assigned to a tier.
|
||||
|
||||
Raises 404 if tier not found, 422 if any feature codes are invalid.
|
||||
"""
|
||||
tier = feature_service.update_tier_features(db, tier_code, request.feature_codes)
|
||||
db.commit()
|
||||
|
||||
logger.info(
|
||||
f"Updated tier {tier_code} features to {len(request.feature_codes)} features "
|
||||
f"by admin {current_user.id}"
|
||||
)
|
||||
|
||||
return TierFeaturesResponse(
|
||||
id=tier.id,
|
||||
code=tier.code,
|
||||
name=tier.name,
|
||||
description=tier.description,
|
||||
features=tier.features or [],
|
||||
feature_count=len(tier.features or []),
|
||||
)
|
||||
|
||||
|
||||
@admin_features_router.get("/tiers/{tier_code}/features", response_model=TierFeatureDetailResponse)
|
||||
def get_tier_features(
|
||||
tier_code: str,
|
||||
current_user: UserContext = Depends(get_current_admin_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get features assigned to a specific tier with full details.
|
||||
|
||||
Raises 404 if tier not found.
|
||||
"""
|
||||
tier, features = feature_service.get_tier_features_with_details(db, tier_code)
|
||||
|
||||
return TierFeatureDetailResponse(
|
||||
tier_code=tier.code,
|
||||
tier_name=tier.name,
|
||||
features=[
|
||||
{
|
||||
"code": f.code,
|
||||
"name": f.name,
|
||||
"category": f.category,
|
||||
"description": f.description,
|
||||
}
|
||||
for f in features
|
||||
],
|
||||
feature_count=len(features),
|
||||
)
|
||||
@@ -223,7 +223,9 @@ def get_invoices(
|
||||
from app.modules.billing.routes.api.vendor_features import vendor_features_router
|
||||
from app.modules.billing.routes.api.vendor_checkout import vendor_checkout_router
|
||||
from app.modules.billing.routes.api.vendor_addons import vendor_addons_router
|
||||
from app.modules.billing.routes.api.vendor_usage import vendor_usage_router
|
||||
|
||||
vendor_router.include_router(vendor_features_router, tags=["vendor-features"])
|
||||
vendor_router.include_router(vendor_checkout_router, tags=["vendor-billing"])
|
||||
vendor_router.include_router(vendor_addons_router, tags=["vendor-billing-addons"])
|
||||
vendor_router.include_router(vendor_usage_router, tags=["vendor-usage"])
|
||||
|
||||
353
app/modules/billing/routes/api/vendor_features.py
Normal file
353
app/modules/billing/routes/api/vendor_features.py
Normal file
@@ -0,0 +1,353 @@
|
||||
# app/modules/billing/routes/api/vendor_features.py
|
||||
"""
|
||||
Vendor features API endpoints.
|
||||
|
||||
Provides feature availability information for the frontend to:
|
||||
- Show/hide UI elements based on tier
|
||||
- Display upgrade prompts for unavailable features
|
||||
- Load feature metadata for dynamic rendering
|
||||
|
||||
Endpoints:
|
||||
- GET /features/available - List of feature codes (for quick checks)
|
||||
- GET /features - Full feature list with availability and metadata
|
||||
- GET /features/{code} - Single feature details with upgrade info
|
||||
- GET /features/categories - List feature categories
|
||||
|
||||
All routes require module access control for the 'billing' module.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.deps import get_current_vendor_api, require_module_access
|
||||
from app.core.database import get_db
|
||||
from app.exceptions import FeatureNotFoundError
|
||||
from app.services.feature_service import feature_service
|
||||
from models.schema.auth import UserContext
|
||||
|
||||
vendor_features_router = APIRouter(
|
||||
prefix="/features",
|
||||
dependencies=[Depends(require_module_access("billing"))],
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Response Schemas
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class FeatureCodeListResponse(BaseModel):
|
||||
"""Simple list of available feature codes for quick checks."""
|
||||
|
||||
features: list[str]
|
||||
tier_code: str
|
||||
tier_name: str
|
||||
|
||||
|
||||
class FeatureResponse(BaseModel):
|
||||
"""Full feature information."""
|
||||
|
||||
code: str
|
||||
name: str
|
||||
description: str | None = None
|
||||
category: str
|
||||
ui_location: str | None = None
|
||||
ui_icon: str | None = None
|
||||
ui_route: str | None = None
|
||||
ui_badge_text: str | None = None
|
||||
is_available: bool
|
||||
minimum_tier_code: str | None = None
|
||||
minimum_tier_name: str | None = None
|
||||
|
||||
|
||||
class FeatureListResponse(BaseModel):
|
||||
"""List of features with metadata."""
|
||||
|
||||
features: list[FeatureResponse]
|
||||
available_count: int
|
||||
total_count: int
|
||||
tier_code: str
|
||||
tier_name: str
|
||||
|
||||
|
||||
class FeatureDetailResponse(BaseModel):
|
||||
"""Single feature detail with upgrade info."""
|
||||
|
||||
code: str
|
||||
name: str
|
||||
description: str | None = None
|
||||
category: str
|
||||
ui_location: str | None = None
|
||||
ui_icon: str | None = None
|
||||
ui_route: str | None = None
|
||||
is_available: bool
|
||||
# Upgrade info (only if not available)
|
||||
upgrade_tier_code: str | None = None
|
||||
upgrade_tier_name: str | None = None
|
||||
upgrade_tier_price_monthly_cents: int | None = None
|
||||
|
||||
|
||||
class CategoryListResponse(BaseModel):
|
||||
"""List of feature categories."""
|
||||
|
||||
categories: list[str]
|
||||
|
||||
|
||||
class FeatureGroupedResponse(BaseModel):
|
||||
"""Features grouped by category."""
|
||||
|
||||
categories: dict[str, list[FeatureResponse]]
|
||||
available_count: int
|
||||
total_count: int
|
||||
|
||||
|
||||
class FeatureCheckResponse(BaseModel):
|
||||
"""Quick feature availability check response."""
|
||||
|
||||
has_feature: bool
|
||||
feature_code: str
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Endpoints
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@vendor_features_router.get("/available", response_model=FeatureCodeListResponse)
|
||||
def get_available_features(
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get list of feature codes available to vendor.
|
||||
|
||||
This is a lightweight endpoint for quick feature checks.
|
||||
Use this to populate a frontend feature store on app init.
|
||||
|
||||
Returns:
|
||||
List of feature codes the vendor has access to
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
# Get subscription for tier info
|
||||
from app.services.subscription_service import subscription_service
|
||||
|
||||
subscription = subscription_service.get_or_create_subscription(db, vendor_id)
|
||||
tier = subscription.tier_obj
|
||||
|
||||
# Get available features
|
||||
feature_codes = feature_service.get_available_feature_codes(db, vendor_id)
|
||||
|
||||
return FeatureCodeListResponse(
|
||||
features=feature_codes,
|
||||
tier_code=subscription.tier,
|
||||
tier_name=tier.name if tier else subscription.tier.title(),
|
||||
)
|
||||
|
||||
|
||||
@vendor_features_router.get("", response_model=FeatureListResponse)
|
||||
def get_features(
|
||||
category: str | None = Query(None, description="Filter by category"),
|
||||
include_unavailable: bool = Query(True, description="Include features not available to vendor"),
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get all features with availability status and metadata.
|
||||
|
||||
This is a comprehensive endpoint for building feature-gated UIs.
|
||||
Each feature includes:
|
||||
- Availability status
|
||||
- UI metadata (icon, route, location)
|
||||
- Minimum tier required
|
||||
|
||||
Args:
|
||||
category: Filter to specific category (orders, inventory, etc.)
|
||||
include_unavailable: Whether to include locked features
|
||||
|
||||
Returns:
|
||||
List of features with metadata and availability
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
# Get subscription for tier info
|
||||
from app.services.subscription_service import subscription_service
|
||||
|
||||
subscription = subscription_service.get_or_create_subscription(db, vendor_id)
|
||||
tier = subscription.tier_obj
|
||||
|
||||
# Get features
|
||||
features = feature_service.get_vendor_features(
|
||||
db,
|
||||
vendor_id,
|
||||
category=category,
|
||||
include_unavailable=include_unavailable,
|
||||
)
|
||||
|
||||
available_count = sum(1 for f in features if f.is_available)
|
||||
|
||||
return FeatureListResponse(
|
||||
features=[
|
||||
FeatureResponse(
|
||||
code=f.code,
|
||||
name=f.name,
|
||||
description=f.description,
|
||||
category=f.category,
|
||||
ui_location=f.ui_location,
|
||||
ui_icon=f.ui_icon,
|
||||
ui_route=f.ui_route,
|
||||
ui_badge_text=f.ui_badge_text,
|
||||
is_available=f.is_available,
|
||||
minimum_tier_code=f.minimum_tier_code,
|
||||
minimum_tier_name=f.minimum_tier_name,
|
||||
)
|
||||
for f in features
|
||||
],
|
||||
available_count=available_count,
|
||||
total_count=len(features),
|
||||
tier_code=subscription.tier,
|
||||
tier_name=tier.name if tier else subscription.tier.title(),
|
||||
)
|
||||
|
||||
|
||||
@vendor_features_router.get("/categories", response_model=CategoryListResponse)
|
||||
def get_feature_categories(
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get list of feature categories.
|
||||
|
||||
Returns:
|
||||
List of category names
|
||||
"""
|
||||
categories = feature_service.get_categories(db)
|
||||
return CategoryListResponse(categories=categories)
|
||||
|
||||
|
||||
@vendor_features_router.get("/grouped", response_model=FeatureGroupedResponse)
|
||||
def get_features_grouped(
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get features grouped by category.
|
||||
|
||||
Useful for rendering feature comparison tables or settings pages.
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
grouped = feature_service.get_features_grouped_by_category(db, vendor_id)
|
||||
|
||||
# Convert to response format
|
||||
categories_response = {}
|
||||
total = 0
|
||||
available = 0
|
||||
|
||||
for category, features in grouped.items():
|
||||
categories_response[category] = [
|
||||
FeatureResponse(
|
||||
code=f.code,
|
||||
name=f.name,
|
||||
description=f.description,
|
||||
category=f.category,
|
||||
ui_location=f.ui_location,
|
||||
ui_icon=f.ui_icon,
|
||||
ui_route=f.ui_route,
|
||||
ui_badge_text=f.ui_badge_text,
|
||||
is_available=f.is_available,
|
||||
minimum_tier_code=f.minimum_tier_code,
|
||||
minimum_tier_name=f.minimum_tier_name,
|
||||
)
|
||||
for f in features
|
||||
]
|
||||
total += len(features)
|
||||
available += sum(1 for f in features if f.is_available)
|
||||
|
||||
return FeatureGroupedResponse(
|
||||
categories=categories_response,
|
||||
available_count=available,
|
||||
total_count=total,
|
||||
)
|
||||
|
||||
|
||||
@vendor_features_router.get("/{feature_code}", response_model=FeatureDetailResponse)
|
||||
def get_feature_detail(
|
||||
feature_code: str,
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get detailed information about a specific feature.
|
||||
|
||||
Includes upgrade information if the feature is not available.
|
||||
Use this for upgrade prompts and feature explanation modals.
|
||||
|
||||
Args:
|
||||
feature_code: The feature code
|
||||
|
||||
Returns:
|
||||
Feature details with upgrade info if locked
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
# Get feature
|
||||
feature = feature_service.get_feature_by_code(db, feature_code)
|
||||
if not feature:
|
||||
raise FeatureNotFoundError(feature_code)
|
||||
|
||||
# Check availability
|
||||
is_available = feature_service.has_feature(db, vendor_id, feature_code)
|
||||
|
||||
# Get upgrade info if not available
|
||||
upgrade_tier_code = None
|
||||
upgrade_tier_name = None
|
||||
upgrade_tier_price = None
|
||||
|
||||
if not is_available:
|
||||
upgrade_info = feature_service.get_feature_upgrade_info(db, feature_code)
|
||||
if upgrade_info:
|
||||
upgrade_tier_code = upgrade_info.required_tier_code
|
||||
upgrade_tier_name = upgrade_info.required_tier_name
|
||||
upgrade_tier_price = upgrade_info.required_tier_price_monthly_cents
|
||||
|
||||
return FeatureDetailResponse(
|
||||
code=feature.code,
|
||||
name=feature.name,
|
||||
description=feature.description,
|
||||
category=feature.category,
|
||||
ui_location=feature.ui_location,
|
||||
ui_icon=feature.ui_icon,
|
||||
ui_route=feature.ui_route,
|
||||
is_available=is_available,
|
||||
upgrade_tier_code=upgrade_tier_code,
|
||||
upgrade_tier_name=upgrade_tier_name,
|
||||
upgrade_tier_price_monthly_cents=upgrade_tier_price,
|
||||
)
|
||||
|
||||
|
||||
@vendor_features_router.get("/check/{feature_code}", response_model=FeatureCheckResponse)
|
||||
def check_feature(
|
||||
feature_code: str,
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Quick check if vendor has access to a feature.
|
||||
|
||||
Returns simple boolean response for inline checks.
|
||||
|
||||
Args:
|
||||
feature_code: The feature code
|
||||
|
||||
Returns:
|
||||
has_feature and feature_code
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
has_feature = feature_service.has_feature(db, vendor_id, feature_code)
|
||||
|
||||
return FeatureCheckResponse(has_feature=has_feature, feature_code=feature_code)
|
||||
182
app/modules/billing/routes/api/vendor_usage.py
Normal file
182
app/modules/billing/routes/api/vendor_usage.py
Normal file
@@ -0,0 +1,182 @@
|
||||
# app/modules/billing/routes/api/vendor_usage.py
|
||||
"""
|
||||
Vendor usage and limits API endpoints.
|
||||
|
||||
Provides endpoints for:
|
||||
- Current usage vs limits
|
||||
- Upgrade recommendations
|
||||
- Approaching limit warnings
|
||||
|
||||
Migrated from app/api/v1/vendor/usage.py to billing module.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.deps import get_current_vendor_api, require_module_access
|
||||
from app.core.database import get_db
|
||||
from app.services.usage_service import usage_service
|
||||
from models.schema.auth import UserContext
|
||||
|
||||
vendor_usage_router = APIRouter(
|
||||
prefix="/usage",
|
||||
dependencies=[Depends(require_module_access("billing"))],
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Response Schemas
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class UsageMetric(BaseModel):
|
||||
"""Single usage metric."""
|
||||
|
||||
name: str
|
||||
current: int
|
||||
limit: int | None # None = unlimited
|
||||
percentage: float # 0-100, or 0 if unlimited
|
||||
is_unlimited: bool
|
||||
is_at_limit: bool
|
||||
is_approaching_limit: bool # >= 80%
|
||||
|
||||
|
||||
class TierInfo(BaseModel):
|
||||
"""Current tier information."""
|
||||
|
||||
code: str
|
||||
name: str
|
||||
price_monthly_cents: int
|
||||
is_highest_tier: bool
|
||||
|
||||
|
||||
class UpgradeTierInfo(BaseModel):
|
||||
"""Next tier upgrade information."""
|
||||
|
||||
code: str
|
||||
name: str
|
||||
price_monthly_cents: int
|
||||
price_increase_cents: int
|
||||
benefits: list[str]
|
||||
|
||||
|
||||
class UsageResponse(BaseModel):
|
||||
"""Full usage response with limits and upgrade info."""
|
||||
|
||||
tier: TierInfo
|
||||
usage: list[UsageMetric]
|
||||
has_limits_approaching: bool
|
||||
has_limits_reached: bool
|
||||
upgrade_available: bool
|
||||
upgrade_tier: UpgradeTierInfo | None = None
|
||||
upgrade_reasons: list[str]
|
||||
|
||||
|
||||
class LimitCheckResponse(BaseModel):
|
||||
"""Response for checking a specific limit."""
|
||||
|
||||
limit_type: str
|
||||
can_proceed: bool
|
||||
current: int
|
||||
limit: int | None
|
||||
percentage: float
|
||||
message: str | None = None
|
||||
upgrade_tier_code: str | None = None
|
||||
upgrade_tier_name: str | None = None
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Endpoints
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@vendor_usage_router.get("", response_model=UsageResponse)
|
||||
def get_usage(
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get current usage, limits, and upgrade recommendations.
|
||||
|
||||
Returns comprehensive usage info for displaying in dashboard
|
||||
and determining when to show upgrade prompts.
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
# Get usage data from service
|
||||
usage_data = usage_service.get_vendor_usage(db, vendor_id)
|
||||
|
||||
# Convert to response
|
||||
return UsageResponse(
|
||||
tier=TierInfo(
|
||||
code=usage_data.tier.code,
|
||||
name=usage_data.tier.name,
|
||||
price_monthly_cents=usage_data.tier.price_monthly_cents,
|
||||
is_highest_tier=usage_data.tier.is_highest_tier,
|
||||
),
|
||||
usage=[
|
||||
UsageMetric(
|
||||
name=m.name,
|
||||
current=m.current,
|
||||
limit=m.limit,
|
||||
percentage=m.percentage,
|
||||
is_unlimited=m.is_unlimited,
|
||||
is_at_limit=m.is_at_limit,
|
||||
is_approaching_limit=m.is_approaching_limit,
|
||||
)
|
||||
for m in usage_data.usage
|
||||
],
|
||||
has_limits_approaching=usage_data.has_limits_approaching,
|
||||
has_limits_reached=usage_data.has_limits_reached,
|
||||
upgrade_available=usage_data.upgrade_available,
|
||||
upgrade_tier=(
|
||||
UpgradeTierInfo(
|
||||
code=usage_data.upgrade_tier.code,
|
||||
name=usage_data.upgrade_tier.name,
|
||||
price_monthly_cents=usage_data.upgrade_tier.price_monthly_cents,
|
||||
price_increase_cents=usage_data.upgrade_tier.price_increase_cents,
|
||||
benefits=usage_data.upgrade_tier.benefits,
|
||||
)
|
||||
if usage_data.upgrade_tier
|
||||
else None
|
||||
),
|
||||
upgrade_reasons=usage_data.upgrade_reasons,
|
||||
)
|
||||
|
||||
|
||||
@vendor_usage_router.get("/check/{limit_type}", response_model=LimitCheckResponse)
|
||||
def check_limit(
|
||||
limit_type: str,
|
||||
current_user: UserContext = Depends(get_current_vendor_api),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Check a specific limit before performing an action.
|
||||
|
||||
Use this before creating orders, products, or inviting team members.
|
||||
|
||||
Args:
|
||||
limit_type: One of "orders", "products", "team_members"
|
||||
|
||||
Returns:
|
||||
Whether the action can proceed and upgrade info if not
|
||||
"""
|
||||
vendor_id = current_user.token_vendor_id
|
||||
|
||||
# Check limit using service
|
||||
check_data = usage_service.check_limit(db, vendor_id, limit_type)
|
||||
|
||||
return LimitCheckResponse(
|
||||
limit_type=check_data.limit_type,
|
||||
can_proceed=check_data.can_proceed,
|
||||
current=check_data.current,
|
||||
limit=check_data.limit,
|
||||
percentage=check_data.percentage,
|
||||
message=check_data.message,
|
||||
upgrade_tier_code=check_data.upgrade_tier_code,
|
||||
upgrade_tier_name=check_data.upgrade_tier_name,
|
||||
)
|
||||
Reference in New Issue
Block a user