0Pricing
FastAPI Backend Development Bootcamp · บทเรียน

การจำกัดอัตราและการป้องกันการใช้งานโดยบอต

ใช้งานการจำกัดอัตราและการควบคุมปริมาณแบบกระจายด้วย Redis เพื่อรองรับการพุ่งสูงและบล็อกไคลเอ็นต์ที่ใช้งานในทางมิชอบ

การจำกัดอัตราและการป้องกันการใช้งานโดยบอต เป็นบทเรียน FastAPI Backend Development Bootcamp ฟรีบน CoddyKit นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน FastAPI Backend Development Bootcamp และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ

Why Distributed Rate Limiting

A single FastAPI worker holding a rate-limit counter in process memory breaks the moment you scale horizontally. With N Uvicorn workers behind a load balancer, an abusive client gets N times the allowance because each worker counts independently.

  • Goal: one shared counter that every worker and every pod sees.
  • Tool: Redis, an atomic in-memory store that all instances connect to.
  • Targets: absorb legitimate traffic spikes while throttling or blocking abusive bots.

Throughout this lesson we build the algorithms in plain Python first, then wire them into FastAPI dependencies.

The Fixed Window Counter

The simplest algorithm: divide time into fixed windows (e.g. 60-second buckets) and increment a counter per client per window. When the counter exceeds the limit, reject the request.

Below is a pure-Python simulation so you can see the mechanics without Redis. Notice the boundary problem: a client can send the full quota at the end of one window and again at the start of the next, briefly doubling the effective rate.

import time

class FixedWindow:
    def __init__(self, limit, window_seconds):
        self.limit = limit
        self.window = window_seconds
        self.buckets = {}

    def allow(self, key, now):
        win = int(now // self.window)
        bucket_key = (key, win)
        count = self.buckets.get(bucket_key, 0) + 1
        self.buckets[bucket_key] = count
        return count <= self.limit

limiter = FixedWindow(limit=3, window_seconds=60)
base = 1000.0
for i in range(5):
    ok = limiter.allow('user:42', base + i)
    print(f'request {i+1}: {"ALLOW" if ok else "BLOCK"}')

The Sliding Window Log

To kill the boundary problem we track a log of timestamps per client and count only the events that fall within the trailing window relative to now. This is exact but memory-heavy: one entry per request.

The pattern below is exactly what we will translate to a Redis sorted set, where the score is the timestamp.

import time
from collections import deque

class SlidingLog:
    def __init__(self, limit, window_seconds):
        self.limit = limit
        self.window = window_seconds
        self.logs = {}

    def allow(self, key, now):
        dq = self.logs.setdefault(key, deque())
        cutoff = now - self.window
        while dq and dq[0] <= cutoff:
            dq.popleft()
        if len(dq) < self.limit:
            dq.append(now)
            return True
        return False

limiter = SlidingLog(limit=2, window_seconds=10)
for t in [0, 1, 2, 11, 12]:
    ok = limiter.allow('ip:1.2.3.4', float(t))
    print(f't={t}s -> {"ALLOW" if ok else "BLOCK"}')

Token Bucket: Absorbing Spikes

Fixed and sliding windows are strict. To absorb bursts while enforcing a long-run average, the token bucket is the standard choice.

  • Each client owns a bucket with a capacity (max burst) and a refill rate (tokens per second).
  • Each request consumes one token; if the bucket is empty, the request is throttled.
  • A burst of up to capacity requests passes instantly, then traffic is smoothed to the refill rate.

This is the algorithm we recommend for public APIs facing spiky-but-legitimate traffic.

class TokenBucket:
    def __init__(self, capacity, refill_per_sec):
        self.capacity = capacity
        self.refill = refill_per_sec
        self.tokens = capacity
        self.last = 0.0

    def allow(self, now):
        elapsed = now - self.last
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill)
        self.last = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

bucket = TokenBucket(capacity=5, refill_per_sec=1)
for t in [0, 0, 0, 0, 0, 0, 3]:
    print(f't={t}: {"ALLOW" if bucket.allow(float(t)) else "THROTTLE"}')

Why Atomicity Matters

Across many workers the read-modify-write of a counter is a classic race condition: two workers read count=4, both increment, both write 5, and a request that should have been blocked slips through.

Redis solves this because each command is atomic, but a rate-limit decision usually needs several commands (increment, set expiry, compare). The robust solution is a Lua script executed by EVAL: Redis runs the whole script atomically, with no other client interleaving.

The next scenes show the Redis-backed fixed window and token bucket using this approach.

Redis Fixed Window with INCR + EXPIRE

The cheapest distributed limiter: a Redis key per client per window. INCR returns the new count atomically; on the first hit we set a TTL equal to the window so the key self-cleans.

We wrap both commands in a small Lua script so the increment and the expiry are one atomic unit. This is FastAPI-side glue code, not a standalone program.

import redis.asyncio as redis

FIXED_WINDOW_LUA = """
local current = redis.call('INCR', KEYS[1])
if current == 1 then
  redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return current
"""

class RedisFixedWindow:
    def __init__(self, client, limit, window):
        self.client = client
        self.limit = limit
        self.window = window
        self.script = client.register_script(FIXED_WINDOW_LUA)

    async def allow(self, identifier: str) -> bool:
        bucket = int(__import__('time').time()) // self.window
        key = f'rl:fw:{identifier}:{bucket}'
        count = await self.script(keys=[key], args=[self.window])
        return count <= self.limit

# rl = RedisFixedWindow(redis.from_url('redis://localhost'), 100, 60)

Redis Token Bucket in Lua

For burst absorption we port the token bucket to a Lua script. State lives in a Redis hash holding tokens and ts (last refill time). The script refills based on elapsed time, tries to consume one token, and writes the state back, all atomically.

Returning 1 means allowed, 0 means throttled. We also set a TTL so idle clients free their memory.

TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local state = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(state[1]) or capacity
local ts = tonumber(state[2]) or now
local delta = math.max(0, now - ts)
tokens = math.min(capacity, tokens + delta * refill)
local allowed = 0
if tokens >= 1 then
  tokens = tokens - 1
  allowed = 1
end
redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill) * 2)
return allowed
"""
# Invoked with EVAL via redis-py: script(keys=[key], args=[cap, refill, time.time()])

A Reusable FastAPI Dependency

We expose the limiter as a dependency so any route can opt in. The dependency derives the client identity, runs the limiter, and raises HTTPException(429) when the quota is exhausted.

Prefer an authenticated identity (API key, user id) over raw IP, because IPs are shared behind NAT and trivially rotated by bots. Fall back to IP only for anonymous traffic.

from fastapi import Request, HTTPException, Depends

def client_identity(request: Request) -> str:
    api_key = request.headers.get('x-api-key')
    if api_key:
        return f'key:{api_key}'
    return f'ip:{request.client.host}'

def rate_limit(limit: int, window: int):
    async def dependency(request: Request):
        ident = client_identity(request)
        limiter = request.app.state.limiter
        if not await limiter.allow(f'{ident}:{limit}:{window}'):
            raise HTTPException(status_code=429, detail='Rate limit exceeded')
    return dependency

# @app.get('/search', dependencies=[Depends(rate_limit(30, 60))])
# async def search(): ...

Honest 429s: Retry-After and Headers

A correct limiter is also a polite one. Well-behaved clients respect standard signals; sending them reduces retries and support tickets.

  • 429 Too Many Requests is the only correct status; never use 403 or 503 for throttling.
  • Retry-After tells the client how many seconds to wait.
  • X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset let clients self-pace.

Have your Lua script also return the remaining count and reset time so you can populate these headers without extra round-trips.

from fastapi import Request, HTTPException

async def enforce(request: Request, limit: int, window: int):
    limiter = request.app.state.limiter
    allowed, remaining, reset_in = await limiter.check(client_identity(request))
    if not allowed:
        raise HTTPException(
            status_code=429,
            detail='Rate limit exceeded',
            headers={
                'Retry-After': str(reset_in),
                'X-RateLimit-Limit': str(limit),
                'X-RateLimit-Remaining': '0',
                'X-RateLimit-Reset': str(reset_in),
            },
        )

Tiered Limits and Abuse Escalation

One global limit is blunt. Production systems layer several:

  • Per-route limits: a cheap GET tolerates far more traffic than an expensive search or login endpoint.
  • Tiered identities: anonymous IPs get a tight quota, authenticated users more, paid plans the most.
  • Escalation: when a client repeatedly hits 429, write it to a Redis denylist with an exponential TTL so abusive bots get progressively longer bans.

The helper below computes a doubling ban duration capped at one hour.

def next_ban_seconds(strikes: int) -> int:
    base = 60  # 1 minute
    cap = 3600  # 1 hour
    return min(cap, base * (2 ** strikes))

for s in range(8):
    print(f'strike {s}: ban for {next_ban_seconds(s)}s')

Detecting Bots Beyond Counting

Rate limiting caps volume, but sophisticated bots stay just under the limit. Combine throttling with cheap behavioral signals:

  • Missing or junk headers: absent User-Agent, or a UA on a known bad list.
  • Login failure ratio: a high failed-to-successful auth ratio per IP signals credential stuffing.
  • Path entropy: rapid hits across many unrelated endpoints suggest scraping.
  • Proof of work / CAPTCHA as a gate when a score crosses a threshold, rather than an outright block.

Feed these into a per-client risk score in Redis and tighten the token-bucket capacity dynamically for high-risk clients.

Quick Check: Choosing the Algorithm

Your public FastAPI API runs across many Uvicorn workers and several pods. Legitimate clients sometimes send short legitimate bursts, but you must enforce a steady long-run average and keep the decision consistent across all instances. Which design fits best?

Recap and Production Checklist

You can now build distributed, abuse-resistant throttling for FastAPI:

  • Centralize state in Redis so every worker and pod shares one counter.
  • Pick the algorithm by need: fixed window for simple caps, sliding log for exactness, token bucket to absorb bursts with an enforced average.
  • Guarantee atomicity with Lua scripts via EVAL to avoid read-modify-write races.
  • Expose limiters as FastAPI dependencies keyed on authenticated identity first, IP as fallback.
  • Respond honestly with 429, Retry-After, and X-RateLimit-* headers.
  • Layer defenses: per-route and per-tier limits, exponential-backoff denylists, and behavioral bot scoring beyond raw counting.

Always fail open carefully: if Redis is unreachable, decide deliberately whether to allow or block, and alert on it.

คำถามที่พบบ่อย

บทเรียน “การจำกัดอัตราและการป้องกันการใช้งานโดยบอต” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การจำกัดอัตราและการป้องกันการใช้งานโดยบอต” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส FastAPI Backend Development Bootcamp ให้อัปเกรดเป็น CoddyKit PRO คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การจำกัดอัตราและการป้องกันการใช้งานโดยบอต”

ใช้งานการจำกัดอัตราและการควบคุมปริมาณแบบกระจายด้วย Redis เพื่อรองรับการพุ่งสูงและบล็อกไคลเอ็นต์ที่ใช้งานในทางมิชอบ คุณปฏิบัติ FastAPI Backend Development Bootcamp ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน FastAPI Backend Development Bootcamp หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน FastAPI Backend Development Bootcamp บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน

บทเรียน “การจำกัดอัตราและการป้องกันการใช้งานโดยบอต” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน FastAPI Backend Development Bootcamp นี้ได้ไหม

ได้ บทเรียน FastAPI Backend Development Bootcamp ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

  1. การลดความเสี่ยงจาก 10 อันดับแรกด้านความปลอดภัย API ของ OWASP
  2. การจำกัดอัตราและการป้องกันการใช้งานโดยบอต
  3. การจัดการข้อมูลลับและการหมุนเวียนคีย์
  4. CORS, CSP และนโยบายส่วนหัวที่ปลอดภัย
← กลับไปที่ FastAPI Backend Development Bootcamp