104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
"""Main routes for the application."""
|
|
|
|
from flask import Blueprint
|
|
|
|
from app.services.decorators import (
|
|
get_current_user,
|
|
require_auth,
|
|
require_credits,
|
|
require_role,
|
|
)
|
|
from app.services.scheduler_service import scheduler_service
|
|
|
|
bp = Blueprint("main", __name__)
|
|
|
|
|
|
@bp.route("/")
|
|
def index() -> dict[str, str]:
|
|
"""Root endpoint that returns API status."""
|
|
return {"message": "API is running", "status": "ok"}
|
|
|
|
|
|
@bp.route("/protected")
|
|
@require_auth
|
|
def protected() -> dict[str, str]:
|
|
"""Protected endpoint that requires authentication."""
|
|
user = get_current_user()
|
|
return {
|
|
"message": f"Hello {user['name']}, this is a protected endpoint!",
|
|
"user": user,
|
|
}
|
|
|
|
|
|
@bp.route("/api-protected")
|
|
@require_auth
|
|
def api_protected() -> dict[str, str]:
|
|
"""Protected endpoint that accepts JWT or API token authentication."""
|
|
user = get_current_user()
|
|
return {
|
|
"message": f"Hello {user['name']}, you accessed this via {user['provider']}!",
|
|
"user": user,
|
|
}
|
|
|
|
|
|
@bp.route("/admin")
|
|
@require_auth
|
|
@require_role("admin")
|
|
def admin_only() -> dict[str, str]:
|
|
"""Admin-only endpoint to demonstrate role-based access."""
|
|
user = get_current_user()
|
|
return {
|
|
"message": f"Hello admin {user['name']}, you have admin access!",
|
|
"user": user,
|
|
"admin_info": "This endpoint is only accessible to admin users",
|
|
}
|
|
|
|
|
|
@bp.route("/health")
|
|
def health() -> dict[str, str]:
|
|
"""Health check endpoint."""
|
|
return {"status": "ok"}
|
|
|
|
|
|
@bp.route("/use-credits/<int:amount>")
|
|
@require_auth
|
|
@require_credits(5)
|
|
def use_credits(amount: int) -> dict[str, str]:
|
|
"""Test endpoint that costs 5 credits to use."""
|
|
user = get_current_user()
|
|
return {
|
|
"message": f"Successfully used endpoint! You requested amount: {amount}",
|
|
"user": user["email"],
|
|
"remaining_credits": user["credits"]
|
|
- 5, # Note: credits already deducted by decorator
|
|
}
|
|
|
|
|
|
@bp.route("/expensive-operation")
|
|
@require_auth
|
|
@require_credits(10)
|
|
def expensive_operation() -> dict[str, str]:
|
|
"""Test endpoint that costs 10 credits to use."""
|
|
user = get_current_user()
|
|
return {
|
|
"message": "Expensive operation completed successfully!",
|
|
"user": user["email"],
|
|
"operation_cost": 10,
|
|
}
|
|
|
|
|
|
@bp.route("/admin/scheduler/status")
|
|
@require_auth
|
|
@require_role("admin")
|
|
def scheduler_status() -> dict:
|
|
"""Get scheduler status (admin only)."""
|
|
return scheduler_service.get_scheduler_status()
|
|
|
|
|
|
@bp.route("/admin/credits/refill", methods=["POST"])
|
|
@require_auth
|
|
@require_role("admin")
|
|
def manual_credit_refill() -> dict:
|
|
"""Manually trigger credit refill for all users (admin only)."""
|
|
return scheduler_service.trigger_credit_refill_now()
|