Spaces:
Runtime error
Runtime error
| """ | |
| DocuLens — Tier & usage API endpoints. | |
| Provides routes for checking usage, listing plans, and managing subscriptions. | |
| """ | |
| import logging | |
| from fastapi import APIRouter, Depends, HTTPException | |
| from middleware import check_rate_limit | |
| from api_usage import ( | |
| get_usage, check_quota, check_feature, get_user_tier, | |
| TIER_FEATURES, TIER_LIMITS, | |
| ) | |
| logger = logging.getLogger(__name__) | |
| tiers_router = APIRouter(tags=["tiers"]) | |
| # --------------------------------------------------------------------------- | |
| # Public — plan information | |
| # --------------------------------------------------------------------------- | |
| async def list_plans(): | |
| """List all available pricing plans and their features.""" | |
| plans = [] | |
| pricing = { | |
| "free": {"price": 0, "label": "Free"}, | |
| "starter": {"price": 29, "label": "Starter"}, | |
| "pro": {"price": 79, "label": "Pro"}, | |
| "enterprise": {"price": 199, "label": "Enterprise"}, | |
| } | |
| for tier_id, features in TIER_FEATURES.items(): | |
| p = pricing.get(tier_id, {"price": 0, "label": tier_id}) | |
| plans.append({ | |
| "id": tier_id, | |
| "name": p["label"], | |
| "price_monthly": p["price"], | |
| "features": features, | |
| }) | |
| return {"plans": plans} | |
| # --------------------------------------------------------------------------- | |
| # Authenticated — user usage & quota | |
| # --------------------------------------------------------------------------- | |
| async def get_user_usage( | |
| user_id: str = "", | |
| api_key: str = Depends(check_rate_limit), | |
| ): | |
| """Get the authenticated user's current month usage and limits.""" | |
| if not user_id: | |
| raise HTTPException(status_code=400, detail="user_id is required") | |
| usage = get_usage(user_id) | |
| return usage | |
| async def check_user_quota( | |
| user_id: str = "", | |
| pages: int = 1, | |
| api_key: str = Depends(check_rate_limit), | |
| ): | |
| """Pre-check whether the user can process the requested number of pages.""" | |
| if not user_id: | |
| raise HTTPException(status_code=400, detail="user_id is required") | |
| result = check_quota(user_id, pages) | |
| if not result["allowed"]: | |
| raise HTTPException(status_code=429, detail=result["message"]) | |
| return result | |
| async def get_tier_features( | |
| user_id: str = "", | |
| api_key: str = Depends(check_rate_limit), | |
| ): | |
| """Get the feature flags for the user's current tier.""" | |
| if not user_id: | |
| raise HTTPException(status_code=400, detail="user_id is required") | |
| tier = get_user_tier(user_id) | |
| features = TIER_FEATURES.get(tier, TIER_FEATURES["free"]) | |
| return {"tier": tier, "features": features} | |