"""Admin routes for the application.""" from flask import Blueprint from app.services.decorators import get_current_user, require_auth, require_role from app.services.scheduler_service import scheduler_service bp = Blueprint("admin", __name__) @bp.route("/") @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("/scheduler/status") @require_auth @require_role("admin") def scheduler_status() -> dict: """Get scheduler status (admin only).""" return scheduler_service.get_scheduler_status() @bp.route("/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() @bp.route("/users") @require_auth @require_role("admin") def list_users() -> dict: """List all users (admin only).""" from app.models.user import User users = User.query.order_by(User.created_at.desc()).all() return { "users": [user.to_dict() for user in users], "total": len(users) } @bp.route("/users/", methods=["PATCH"]) @require_auth @require_role("admin") def update_user(user_id: int) -> dict: """Update user information (admin only).""" from flask import request from app.database import db from app.models.user import User from app.models.plan import Plan data = request.get_json() if not data: return {"error": "No data provided"}, 400 user = User.query.get(user_id) if not user: return {"error": "User not found"}, 404 # Validate and update fields try: if "name" in data: name = data["name"].strip() if not name: return {"error": "Name cannot be empty"}, 400 if len(name) > 100: return {"error": "Name too long (max 100 characters)"}, 400 user.name = name if "credits" in data: credits = data["credits"] if not isinstance(credits, int) or credits < 0: return {"error": "Credits must be a non-negative integer"}, 400 user.credits = credits if "plan_id" in data: plan_id = data["plan_id"] if not isinstance(plan_id, int): return {"error": "Plan ID must be an integer"}, 400 plan = Plan.query.get(plan_id) if not plan: return {"error": "Plan not found"}, 404 user.plan_id = plan_id if "is_active" in data: is_active = data["is_active"] if not isinstance(is_active, bool): return {"error": "is_active must be a boolean"}, 400 user.is_active = is_active db.session.commit() return { "message": "User updated successfully", "user": user.to_dict() } except Exception as e: db.session.rollback() return {"error": f"Failed to update user: {str(e)}"}, 500 @bp.route("/users//deactivate", methods=["POST"]) @require_auth @require_role("admin") def deactivate_user(user_id: int) -> dict: """Deactivate a user (admin only).""" from app.database import db from app.models.user import User user = User.query.get(user_id) if not user: return {"error": "User not found"}, 404 # Prevent admin from deactivating themselves current_user = get_current_user() if str(user.id) == current_user["id"]: return {"error": "Cannot deactivate your own account"}, 400 try: user.deactivate() db.session.commit() return { "message": "User deactivated successfully", "user": user.to_dict() } except Exception as e: db.session.rollback() return {"error": f"Failed to deactivate user: {str(e)}"}, 500 @bp.route("/users//activate", methods=["POST"]) @require_auth @require_role("admin") def activate_user(user_id: int) -> dict: """Activate a user (admin only).""" from app.database import db from app.models.user import User user = User.query.get(user_id) if not user: return {"error": "User not found"}, 404 try: user.activate() db.session.commit() return { "message": "User activated successfully", "user": user.to_dict() } except Exception as e: db.session.rollback() return {"error": f"Failed to activate user: {str(e)}"}, 500 @bp.route("/plans") @require_auth @require_role("admin") def list_plans() -> dict: """List all available plans (admin only).""" from app.models.plan import Plan plans = Plan.query.order_by(Plan.id).all() return { "plans": [plan.to_dict() for plan in plans], "total": len(plans) }