Add /api/v1/usage dashboard endpoint

{
  "cost": {
    "input_tokens": 54,
    "estimated_usd": 0.059157,
    "output_tokens": 3933
  },
  "model": "claude-sonnet-4-20250514",
  "status": "success",
  "transport": "anthropic_api",
  "latency_ms": 42199,
  "response_text": "I'll help you create a usage dashboard endpoint that returns current period usage statistics with API key authentication.\n\n```python\n# app/api/v1/endpoints/usage.py\nfrom datetime import datetime, timezone\nfrom typing import Dict, Any\n\nfrom fastapi import APIRouter, Depends, HTTPException, status\nfrom sqlalchemy.orm import Session\n\nfrom app.core.database import get_db\nfrom app.core.auth import get_current_user_from_api_key\nfrom app.models.user import User\nfrom app.models.usage import Usage\nfrom app.models.subscription import Subscription\nfrom app.schemas.usage import UsageStatsResponse\n\nrouter = APIRouter()\n\n@router.get(\"/usage\", response_model=UsageStatsResponse)\nasync def get_usage_stats(\n    current_user: User = Depends(get_current_user_from_api_key),\n    db: Session = Depends(get_db)\n) -> UsageStatsResponse:\n    \"\"\"\n    Get current period usage statistics for the authenticated user.\n    \n    Returns:\n    - calls_used: Number of API calls made in current period\n    - calls_remaining: Number of calls remaining in current period\n    - period_start: Start date of current billing period\n    - period_end: End date of current billing period\n    - tier_name: Current subscription tier name\n    \"\"\"\n    try:\n        # Get user's current subscription\n        subscription = db.query(Subscription).filter(\n            Subscription.user_id == current_user.id,\n            Subscription.is_active == True\n        ).first()\n        \n        if not subscription:\n            raise HTTPException(\n                status_code=status.HTTP_404_NOT_FOUND,\n                detail=\"No active subscription found\"\n            )\n        \n        # Get current period dates\n        now = datetime.now(timezone.utc)\n        period_start = subscription.current_period_start\n        period_end = subscription.current_period_end\n        \n        # Get usage stats for current period\n        usage_record = db.query(Usage).filter(\n            Usage.user_id == current_user.id,\n            Usage.period_start == period_start,\n            Usage.period_end == period_end\n        ).first()\n        \n        # Calculate usage statistics\n        calls_used = usage_record.calls_made if usage_record else 0\n        calls_limit = subscription.tier.call_limit\n        calls_remaining = max(0, calls_limit - calls_used)\n        \n        return UsageStatsResponse(\n            calls_used=calls_used,\n            calls_remaining=calls_remaining,\n            calls_limit=calls_limit,\n            period_start=period_start,\n            period_end=period_end,\n            tier_name=subscription.tier.name,\n            usage_percentage=round((calls_used / calls_limit) * 100, 2) if calls_limit > 0 else 0\n        )\n        \n    except Exception as e:\n        raise HTTPException(\n            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,\n            detail=f\"Error retrieving usage stats: {str(e)}\"\n        )\n```\n\n```python\n# app/schemas/usage.py\nfrom datetime import datetime\nfrom pydantic import BaseModel, Field\n\nclass UsageStatsResponse(BaseModel):\n    \"\"\"Response schema for usage statistics\"\"\"\n    \n    calls_used: int = Field(..., description=\"Number of API calls made in current period\")\n    calls_remaining: int = Field(..., description=\"Number of calls remaining in current period\")\n    calls_limit: int = Field(..., description=\"Total call limit for current tier\")\n    period_start: datetime = Field(..., description=\"Start date of current billing period\")\n    period_end: datetime = Field(..., description=\"End date of current billing period\")\n    tier_name: str = Field(..., description=\"Current subscription tier name\")\n    usage_percentage: float = Field(..., description=\"Percentage of calls used (0-100)\")\n    \n    class Config:\n        from_attributes = True\n        json_encoders = {\n            datetime: lambda v: v.isoformat()\n        }\n\nclass UsageStatsRequest(BaseModel):\n    \"\"\"Optional request schema for filtering usage stats\"\"\"\n    \n    include_history: bool = Field(default=False, description=\"Include historical usage data\")\n    days_back: int = Field(default=30, ge=1, le=365, description=\"Number of days of history to include\")\n```\n\n```python\n# app/core/auth.py\nfrom fastapi import Depends, HTTPException, status\nfrom fastapi.security import HTTPBearer, HTTPAuthorizationCredentials\nfrom sqlalchemy.orm import Session\n\nfrom app.core.database import get_db\nfrom app.models.user import User\nfrom app.models.api_key import ApiKey\n\nsecurity = HTTPBearer()\n\nasync def get_current_user_from_api_key(\n    credentials: HTTPAuthorizationCredentials = Depends(security),\n    db: Session = Depends(get_db)\n) -> User:\n    \"\"\"\n    Authenticate user via API key from Authorization header.\n    Expected format: Bearer <api_key>\n    \"\"\"\n    if not credentials or not credentials.credentials:\n        raise HTTPException(\n            status_code=status.HTTP_401_UNAUTHORIZED,\n            detail=\"API key required\",\n            headers={\"WWW-Authenticate\": \"Bearer\"},\n        )\n    \n    # Query for valid API key\n    api_key_record = db.query(ApiKey).filter(\n        ApiKey.key == credentials.credentials,\n        ApiKey.is_active == True\n    ).first()\n    \n    if not api_key_record:\n        raise HTTPException(\n            status_code=status.HTTP_401_UNAUTHORIZED,\n            detail=\"Invalid API key\",\n            headers={\"WWW-Authenticate\": \"Bearer\"},\n        )\n    \n    # Check if API key is expired\n    if api_key_record.expires_at and api_key_record.expires_at < datetime.now(timezone.utc):\n        raise HTTPException(\n            status_code=status.HTTP_401_UNAUTHORIZED,\n            detail=\"API key expired\",\n            headers={\"WWW-Authenticate\": \"Bearer\"},\n        )\n    \n    # Get associated user\n    user = db.query(User).filter(User.id == api_key_record.user_id).first()\n    \n    if not user or not user.is_active:\n        raise HTTPException(\n            status_code=status.HTTP_401_UNAUTHORIZED,\n            detail=\"User account inactive\",\n        )\n    \n    # Update last used timestamp\n    api_key_record.last_used_at = datetime.now(timezone.utc)\n    db.commit()\n    \n    return user\n```\n\n```python\n# app/models/api_key.py\nfrom datetime import datetime, timezone\nfrom sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey\nfrom sqlalchemy.orm import relationship\n\nfrom app.core.database import Base\n\nclass ApiKey(Base):\n    \"\"\"API Key model for authentication\"\"\"\n    \n    __tablename__ = \"api_keys\"\n    \n    id = Column(Integer, primary_key=True, index=True)\n    user_id = Column(Integer, ForeignKey(\"users.id\"), nullable=False)\n    key = Column(String(64), unique=True, nullable=False, index=True)\n    name = Column(String(100), nullable=True)  # Optional name for the key\n    is_active = Column(Boolean, default=True, nullable=False)\n    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))\n    expires_at = Column(DateTime(timezone=True), nullable=True)\n    last_used_at = Column(DateTime(timezone=True), nullable=True)\n    \n    # Relationships\n    user = relationship(\"User\", back_populates=\"api_keys\")\n```\n\n```python\n# app/models/usage.py\nfrom datetime import datetime, timezone\nfrom sqlalchemy import Column, Integer, DateTime, ForeignKey\nfrom sqlalchemy.orm import relationship\n\nfrom app.core.database import Base\n\nclass Usage(Base):\n    \"\"\"Usage tracking model\"\"\"\n    \n    __tablename__ = \"usage\"\n    \n    id = Column(Integer, primary_key=True, index=True)\n    user_id = Column(Integer, ForeignKey(\"users.id\"), nullable=False)\n    calls_made = Column(Integer, default=0, nullable=False)\n    period_start = Column(DateTime(timezone=True), nullable=False)\n    period_end = Column(DateTime(timezone=True), nullable=False)\n    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))\n    updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))\n    \n    # Relationships\n    user = relationship(\"User\", back_populates=\"usage_records\")\n```\n\n```python\n# app/models/subscription.py\nfrom datetime import datetime, timezone\nfrom sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Numeric\nfrom sqlalchemy.orm import relationship\n\nfrom app.core.database import Base\n\nclass Subscription(Base):\n    \"\"\"User subscription model\"\"\"\n    \n    __tablename__ = \"subscriptions\"\n    \n    id = Column(Integer, primary_key=True, index=True)\n    user_id = Column(Integer, ForeignKey(\"users.id\"), nullable=False)\n    tier_id = Column(Integer, ForeignKey(\"tiers.id\"), nullable=False)\n    is_active = Column(Boolean, default=True, nullable=False)\n    current_period_start = Column(DateTime(timezone=True), nullable=False)\n    current_period_end = Column(DateTime(timezone=True), nullable=False)\n    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))\n    updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))\n    \n    # Relationships\n    user = relationship(\"User\", back_populates=\"subscriptions\")\n    tier = relationship(\"Tier\", back_populates=\"subscriptions\")\n\nclass Tier(Base):\n    \"\"\"Subscription tier model\"\"\"\n    \n    __tablename__ = \"tiers\"\n    \n    id = Column(Integer, primary_key=True, index=True)\n    name = Column(String(50), unique=True, nullable=False)\n    call_limit = Column(Integer, nullable=False)\n    price = Column(Numeric(10, 2), nullable=False)\n    description = Column(String(255), nullable=True)\n    is_active = Column(Boolean, default=True, nullable=False)\n    \n    # Relationships\n    subscriptions = relationship(\"Subscription\", back_populates=\"tier\")\n```\n\n```python\n# app/api/v1/api.py\nfrom fastapi import APIRouter\n\nfrom app.api.v1.endpoints import usage\n\napi_router = APIRouter()\n\n# Include usage endpoints\napi_router.include_router(usage.router, tags=[\"usage\"])\n```\n\n```python\n# tests/test_usage_endpoint.py\nimport pytest\nfrom datetime import datetime, timezone, timedelta\nfrom fastapi.testclient import TestClient\nfrom sqlalchemy.orm import Session\n\nfrom app.main import app\nfrom app.models.user import User\nfrom app.models.api_key import ApiKey\nfrom app.models.subscription import Subscription, Tier\nfrom app.models.usage import Usage\n\nclient = TestClient(app)\n\ndef test_get_usage_stats_success(db_session: Session):\n    \"\"\"Test successful usage stats retrieval\"\"\"\n    \n    # Create test data\n    user = User(email=\"test@example.com\", is_active=True)\n    db_session.add(user)\n    db_session.flush()\n    \n    api_key = ApiKey(\n        user_id=user.id,\n        key=\"test_api_key_12345\",\n        is_active=True\n    )\n    db_session.add(api_key)\n    \n    tier = Tier(name=\"Basic\", call_limit=1000, price=10.00)\n    db_session.add(tier)\n    db_session.flush()\n    \n    now = datetime.now(timezone.utc)\n    subscription = Subscription(\n        user_id=user.id,\n        tier_id=tier.id,\n        is_active=True,\n        current_period_start=now - timedelta(days=15),\n        current_period_end=now + timedelta(days=15)\n    )\n    db_session.add(subscription)\n    \n    usage = Usage(\n        user_id=user.id,\n        calls_made=250,\n        period_start=subscription.current_period_start,\n        period_end=subscription.current_period_end\n    )\n    db_session.add(usage)\n    db_session.commit()\n    \n    # Test request\n    headers = {\"Authorization\": f\"Bearer {api_key.key}\"}\n    response = client.get(\"/api/v1/usage\", headers=headers)\n    \n    assert response.status_code == 200\n    data = response.json()\n    \n    assert data[\"calls_used\"] == 250\n    assert data[\"calls_remaining\"] == 750\n    assert data[\"calls_limit\"] == 1000\n    assert data[\"tier_name\"] == \"Basic\"\n    assert data[\"usage_percentage\"] == 25.0\n\ndef test_get_usage_stats_no_api_key(db_session: Session):\n    \"\"\"Test usage stats request without API key\"\"\"\n    \n    response = client.get(\"/api/v1/usage\")\n    assert response.status_code == 403\n\ndef test_get_usage_stats_invalid_api_key(db_session: Session):\n    \"\"\"Test usage stats request with invalid API key\"\"\"\n    \n    headers = {\"Authorization\": \"Bearer invalid_key\"}\n    response = client.get(\"/api/v1/usage\", headers=headers)\n    \n    assert response.status_code == 401\n    assert \"Invalid API key\" in response.json()[\"detail\"]\n\ndef test_get_usage_stats_no_subscription(db_session: Session):\n    \"\"\"Test usage stats for user without active subscription\"\"\"\n    \n    user = User(email=\"test@example.com\", is_active=True)\n    db_session.add(user)\n    db_session.flush()\n    \n    api_key = ApiKey(\n        user_id=user.id,\n        key=\"test_api_key_12345\",\n        is_active=True\n    )\n    db_session.add(api_key)\n    db_session.commit()\n    \n    headers = {\"Authorization\": f\"Bearer {api_key.key}\"}\n    response = client.get(\"/api/v1/usage\", headers=headers)\n    \n    assert response.status_code == 404\n    assert \"No active subscription found\" in response.json()[\"detail\"]\n```\n\n```bash\n# Example API request\ncurl -X GET \"http://localhost:8000/api/v1/usage\" \\\n  -H \"Authorization: Bearer your_api_key_here\" \\\n  -H \"Content-Type: application/json\"\n\n# Example response\n{\n  \"calls_used\": 250,\n  \"calls_remaining\": 750,\n  \"calls_limit\": 1000,\n  \"period_start\": \"2024-01-01T00:00:00Z\",\n  \"period_end\": \"2024-01-31T23:59:59Z\",\n  \"tier_name\": \"Basic\",\n  \"usage_percentage\": 25.0\n}\n```\n\nThis implementation provides:\n\n1. **GET /api/v1/usage endpoint** that returns comprehensive usage statistics\n2. **API key authentication** via Authorization header (Bearer token)\n3. **Current period tracking** with start/end dates\n4. **Usage calculations** including used, remaining, and percentage\n5. **Error handling** for missing subscriptions and invalid keys\n6. **Comprehensive testing** covering success and error cases\n7. **Type safety** with Pydantic schemas\n8. **Database models** for API keys, usage, subscriptions, and tiers\n\nThe endpoint returns all the requested information: calls used/remaining, period dates, and tier name, with proper authentication required."
}