Spaces:
Runtime error
Runtime error
File size: 2,794 Bytes
04c4194 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """
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
# ---------------------------------------------------------------------------
@tiers_router.get("/api/v1/plans")
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
# ---------------------------------------------------------------------------
@tiers_router.get("/api/v1/usage")
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
@tiers_router.get("/api/v1/usage/check")
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
@tiers_router.get("/api/v1/usage/features")
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}
|