Samir Boulahtit
9a0dd84035
The require_module_access dependency was using path-based detection to determine admin vs vendor authentication, which failed for API routes (/api/v1/admin/*) because it only checked for /admin/*. Changes: - Make frontend_type parameter mandatory (was optional with fallback) - Remove path-based detection logic from require_module_access - Update all 33 module route files to pass explicit FrontendType: - 15 admin routes use FrontendType.ADMIN - 18 vendor routes use FrontendType.VENDOR This ensures authentication method is explicitly declared at route definition time, making it independent of URL structure and future-proof for API version changes. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
# app/modules/loyalty/routes/api/admin.py
|
|
"""
|
|
Loyalty module admin routes.
|
|
|
|
Platform admin endpoints for:
|
|
- Viewing all loyalty programs
|
|
- Platform-wide analytics
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
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.modules.enums import FrontendType
|
|
from app.modules.loyalty.schemas import (
|
|
ProgramListResponse,
|
|
ProgramResponse,
|
|
ProgramStatsResponse,
|
|
)
|
|
from app.modules.loyalty.services import program_service
|
|
from app.modules.tenancy.models import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Admin router with module access control
|
|
admin_router = APIRouter(
|
|
prefix="/loyalty",
|
|
dependencies=[Depends(require_module_access("loyalty", FrontendType.ADMIN))],
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Program Management
|
|
# =============================================================================
|
|
|
|
|
|
@admin_router.get("/programs", response_model=ProgramListResponse)
|
|
def list_programs(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
is_active: bool | None = Query(None),
|
|
current_user: User = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""List all loyalty programs (platform admin)."""
|
|
programs, total = program_service.list_programs(
|
|
db,
|
|
skip=skip,
|
|
limit=limit,
|
|
is_active=is_active,
|
|
)
|
|
|
|
program_responses = []
|
|
for program in programs:
|
|
response = ProgramResponse.model_validate(program)
|
|
response.is_stamps_enabled = program.is_stamps_enabled
|
|
response.is_points_enabled = program.is_points_enabled
|
|
response.display_name = program.display_name
|
|
program_responses.append(response)
|
|
|
|
return ProgramListResponse(programs=program_responses, total=total)
|
|
|
|
|
|
@admin_router.get("/programs/{program_id}", response_model=ProgramResponse)
|
|
def get_program(
|
|
program_id: int,
|
|
current_user: User = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get a specific loyalty program."""
|
|
program = program_service.require_program(db, program_id)
|
|
|
|
response = ProgramResponse.model_validate(program)
|
|
response.is_stamps_enabled = program.is_stamps_enabled
|
|
response.is_points_enabled = program.is_points_enabled
|
|
response.display_name = program.display_name
|
|
|
|
return response
|
|
|
|
|
|
@admin_router.get("/programs/{program_id}/stats", response_model=ProgramStatsResponse)
|
|
def get_program_stats(
|
|
program_id: int,
|
|
current_user: User = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get statistics for a loyalty program."""
|
|
stats = program_service.get_program_stats(db, program_id)
|
|
return ProgramStatsResponse(**stats)
|
|
|
|
|
|
# =============================================================================
|
|
# Platform Stats
|
|
# =============================================================================
|
|
|
|
|
|
@admin_router.get("/stats")
|
|
def get_platform_stats(
|
|
current_user: User = Depends(get_current_admin_api),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Get platform-wide loyalty statistics."""
|
|
from sqlalchemy import func
|
|
|
|
from app.modules.loyalty.models import LoyaltyCard, LoyaltyProgram, LoyaltyTransaction
|
|
|
|
# Program counts
|
|
total_programs = db.query(func.count(LoyaltyProgram.id)).scalar() or 0
|
|
active_programs = (
|
|
db.query(func.count(LoyaltyProgram.id))
|
|
.filter(LoyaltyProgram.is_active == True)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
# Card counts
|
|
total_cards = db.query(func.count(LoyaltyCard.id)).scalar() or 0
|
|
active_cards = (
|
|
db.query(func.count(LoyaltyCard.id))
|
|
.filter(LoyaltyCard.is_active == True)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
# Transaction counts (last 30 days)
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
thirty_days_ago = datetime.now(UTC) - timedelta(days=30)
|
|
transactions_30d = (
|
|
db.query(func.count(LoyaltyTransaction.id))
|
|
.filter(LoyaltyTransaction.transaction_at >= thirty_days_ago)
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
return {
|
|
"total_programs": total_programs,
|
|
"active_programs": active_programs,
|
|
"total_cards": total_cards,
|
|
"active_cards": active_cards,
|
|
"transactions_30d": transactions_30d,
|
|
}
|