+{"cells": [{"cell_type": "markdown", "id": "302934307671667531413257853548643485645", "metadata": {}, "source": ["# Gradio Demo: rate_limit"]}, {"cell_type": "code", "execution_count": null, "id": "272996653310673477252411125948039410165", "metadata": {}, "outputs": [], "source": ["!pip install -q gradio "]}, {"cell_type": "code", "execution_count": null, "id": "288918539441861185822528903084949547379", "metadata": {}, "outputs": [], "source": ["import gradio as gr\n", "from datetime import datetime, timedelta\n", "from collections import defaultdict\n", "import threading\n", "\n", "rate_limit_data = defaultdict(list)\n", "lock = threading.Lock()\n", "\n", "UNAUTH_RATE_LIMIT = 3\n", "AUTH_RATE_LIMIT = 30\n", "RATE_LIMIT_WINDOW = 60\n", "\n", "def clean_old_entries(user_id):\n", " \"\"\"Remove entries older than the rate limit window\"\"\"\n", " current_time = datetime.now()\n", " cutoff_time = current_time - timedelta(seconds=RATE_LIMIT_WINDOW)\n", " rate_limit_data[user_id] = [\n", " timestamp for timestamp in rate_limit_data[user_id]\n", " if timestamp > cutoff_time\n", " ]\n", "\n", "def get_user_identifier(profile: gr.OAuthProfile | None, request: gr.Request) -> tuple[str, bool]:\n", " \"\"\"Get user identifier and whether they're authenticated\"\"\"\n", " if profile is not None:\n", " return profile.username, True\n", " else:\n", " if request:\n", " return f\"ip_{request.client.host}\", False\n", " return \"ip_unknown\", False\n", "\n", "def check_rate_limit(user_id: str, is_authenticated: bool) -> tuple[bool, int, int]:\n", " \"\"\"\n", " Check if user has exceeded rate limit\n", " Returns: (can_proceed, clicks_used, max_clicks)\n", " \"\"\"\n", " with lock:\n", " clean_old_entries(user_id)\n", " \n", " max_clicks = AUTH_RATE_LIMIT if is_authenticated else UNAUTH_RATE_LIMIT\n", " clicks_used = len(rate_limit_data[user_id])\n", " \n", " can_proceed = clicks_used < max_clicks\n", " \n", " return can_proceed, clicks_used, max_clicks\n", "\n", "def add_click(user_id: str):\n", " \"\"\"Add a click timestamp for the user\"\"\"\n", " with lock:\n", " rate_limit_data[user_id].append(datetime.now())\n", "\n", "def update_status(profile: gr.OAuthProfile | None, request: gr.Request) -> str:\n", " \"\"\"Update the status message showing current rate limit info\"\"\"\n", " user_id, is_authenticated = get_user_identifier(profile, request)\n", " _, clicks_used, max_clicks = check_rate_limit(user_id, is_authenticated)\n", " \n", " if is_authenticated:\n", " return f\"\u2705 You are logged in as '{profile.username}'. You have clicked {clicks_used} times this minute. You have {max_clicks} total clicks per minute.\" # type: ignore\n", " else:\n", " return f\"\u26a0\ufe0f You are not logged in. You have clicked {clicks_used} times this minute. You have {max_clicks} total clicks per minute.\"\n", "\n", "def run_action(profile: gr.OAuthProfile | None, request: gr.Request) -> tuple[str, str]:\n", " \"\"\"Handle the run button click with rate limiting\"\"\"\n", " user_id, is_authenticated = get_user_identifier(profile, request)\n", " can_proceed, clicks_used, max_clicks = check_rate_limit(user_id, is_authenticated)\n", " \n", " if not can_proceed:\n", " result = f\"\u274c Rate limit exceeded! You've used all {max_clicks} clicks for this minute. Please wait before trying again.\"\n", " status = update_status(profile, request)\n", " return result, status\n", " \n", " add_click(user_id)\n", " \n", " _, new_clicks_used, _ = check_rate_limit(user_id, is_authenticated)\n", " \n", " result = f\"\u2705 Action executed successfully! (Click #{new_clicks_used})\"\n", " status = update_status(profile, request)\n", " \n", " return result, status\n", "\n", "with gr.Blocks(title=\"Rate Limiting Demo\") as demo:\n", " gr.Markdown(\"# Rate Limiting Demo App\")\n", " gr.Markdown(\"This app demonstrates rate limiting based on authentication status.\")\n", " \n", " gr.LoginButton()\n", " \n", " status_text = gr.Markdown(\"Loading status...\")\n", " \n", " with gr.Row():\n", " run_btn = gr.Button(\"\ud83d\ude80 Run Action\", variant=\"primary\", scale=1)\n", " \n", " result_text = gr.Markdown(\"\")\n", " \n", " demo.load(update_status, inputs=None, outputs=status_text)\n", " \n", " run_btn.click(\n", " run_action,\n", " inputs=None,\n", " outputs=[result_text, status_text]\n", " )\n", " \n", " gr.Markdown(\"---\")\n", " gr.Markdown(\"\"\"\n", " ### Rate Limits:\n", " - **Not logged in:** 3 clicks per minute (based on IP address)\n", " - **Logged in:** 30 clicks per minute (based on HF username)\n", " \n", " ### How it works:\n", " - Click the **Login** button to authenticate with Hugging Face\n", " - Click the **Run Action** button to test the rate limiting\n", " - The system tracks your clicks over a rolling 1-minute window\n", " \"\"\")\n", "\n", "if __name__ == \"__main__\":\n", " demo.launch()\n"]}], "metadata": {}, "nbformat": 4, "nbformat_minor": 5}
0 commit comments