To implement rate limiting in a Streamlit application, you can use a combination of techniques depending on your needs. Below are approaches for different scenarios:

1. Rate Limit Based on User Sessions

Streamlit supports session-based states, and you can use the session state to track user activity.

Code Example:

import streamlit as st
import time

# Initialize session state for rate limiting
if 'last_request_time' not in st.session_state:
    st.session_state.last_request_time = 0

# Set rate limit interval in seconds
RATE_LIMIT = 10

# Check time difference between requests
current_time = time.time()
time_since_last_request = current_time - st.session_state.last_request_time

if time_since_last_request < RATE_LIMIT:
    st.warning(f"Rate limit exceeded. Please wait {RATE_LIMIT - time_since_last_request:.1f} seconds.")
else:
    st.session_state.last_request_time = current_time
    # Add your main application logic here
    st.write("Request processed!")

2. Rate Limit Based on IP Address

For IP-based rate limiting, you can use third-party libraries like Flask-Limiter with Streamlit, or store IP addresses in a database.

Code Example with streamlit.web.Request:

import streamlit as st
from streamlit.runtime.scriptrunner import get_script_run_ctx

# Store IPs and timestamps in a dictionary
ip_requests = {}

# Get client IP
ctx = get_script_run_ctx()
client_ip = ctx.request.client.host if ctx else "unknown"

# Set rate limit interval and max requests
RATE_LIMIT = 10  # seconds
MAX_REQUESTS = 5

# Initialize or update IP tracking
if client_ip not in ip_requests:
    ip_requests[client_ip] = []

current_time = time.time()
ip_requests[client_ip] = [t for t in ip_requests[client_ip] if current_time - t < RATE_LIMIT]

if len(ip_requests[client_ip]) >= MAX_REQUESTS:
    st.warning("Rate limit exceeded. Please try again later.")
else:
    ip_requests[client_ip].append(current_time)
    # Main application logic
    st.write("Request processed for IP:", client_ip)

3. Use a Token Bucket System

Implement a token bucket system to allow users a specific number of requests within a time frame.

Code Example:

import streamlit as st
import time

# Initialize session state for tokens
if 'tokens' not in st.session_state:
    st.session_state.tokens = 5  # Max tokens
    st.session_state.last_refill = time.time()

# Refill tokens based on time
TOKEN_REFILL_TIME = 10  # seconds
TOKEN_MAX = 5

current_time = time.time()
time_since_last_refill = current_time - st.session_state.last_refill
if time_since_last_refill >= TOKEN_REFILL_TIME:
    refill_count = int(time_since_last_refill / TOKEN_REFILL_TIME)
    st.session_state.tokens = min(TOKEN_MAX, st.session_state.tokens + refill_count)
    st.session_state.last_refill = current_time

if st.session_state.tokens > 0:
    st.session_state.tokens -= 1
    # Main application logic
    st.write(f"Request processed! Remaining tokens: {st.session_state.tokens}")
else:
    st.warning("Rate limit exceeded. Please wait for tokens to replenish.")

4. External Tools for Advanced Rate Limiting

For advanced use cases, consider external tools or middleware: - Redis: Store and manage request counts per user or IP. - API Gateway: Use cloud tools like AWS API Gateway, Cloudflare, or Fastly to enforce rate limits.

Redis Example:

import redis
import streamlit as st
import time

# Connect to Redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

RATE_LIMIT = 5
WINDOW_SIZE = 60  # seconds

client_id = "user_123"  # Replace with user ID or IP

# Increment request count
requests = redis_client.incr(client_id)
if requests == 1:
    redis_client.expire(client_id, WINDOW_SIZE)

if requests > RATE_LIMIT:
    st.warning(f"Rate limit exceeded. Try again in {redis_client.ttl(client_id)} seconds.")
else:
    st.write(f"Request processed! Remaining requests: {RATE_LIMIT - requests}")

5. Combine Rate Limiting with Streamlit Cache

Use st.cache or st.cache_data to reduce the frequency of expensive operations like API calls.

Example:

@st.cache(ttl=10)  # Cache data for 10 seconds
def expensive_operation():
    return "Expensive operation result!"

st.write(expensive_operation())

Best Practices

This ensures your Streamlit app remains responsive and prevents abuse.