Fastapi & Rate Limiting

Let’s test a simple Rate Limiting Function found on the Net …

main.py

Main App with Rate Limiting Function

# main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from Token_bucket import TokenBucket

app = FastAPI()

class RateLimiterMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, bucket: TokenBucket):
        super().__init__(app)
        self.bucket = bucket

    async def dispatch(self, request: Request, call_next):
        if self.bucket.take_token():
            return await call_next(request)

        # Return a JSON response for rate limit exceeded
        return JSONResponse(status_code=429, content={"detail": "Rate limit exceeded"})

# Token bucket with a capacity of 3 and refill rate of 1 token per second
bucket = TokenBucket(capacity=3, refill_rate=3)

# Apply the rate limiter middleware
app.add_middleware(RateLimiterMiddleware, bucket=bucket)

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

Token_bucket.py

# Token_bucket.py
import time

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.time()

    def add_tokens(self):
        now = time.time()
        if self.tokens < self.capacity:
            tokens_to_add = (now - self.last_refill) * self.refill_rate
            self.tokens = min (self.capacity, self.tokens + tokens_to_add)
        self.last_refill=now

    def take_token(self):
        self.add_tokens()
        if self.tokens >= 1:
            self.tokens -=1
            return True
        return False

Test Script

Let’s produce some requests …

Project Template for FastAPI

gave a try with a FastAPI Template, https://github.com/rochacbruno/fastapi-project-template.git

Projectname: gugus1234

clone the repo

git clone https://github.com/rochacbruno/fastapi-project-template.git gugus1234
cd gugus1234

Switch Poetry

i’d like to have poetry as virtual env manager

make switch-to-poetry

Rename some Stuff

had to rename some string in pyproject and different files …

mv project_name gugug1234
gsed -i 's/a-flask-test/gugus1234/' pyproject.toml
gsed -i 's/project_name/gugus1234/' pyproject.toml
gsed -i 's/project_name/gugus1234/g' gugus1234/cli.py gugus1234/app.py gugus1234/config.py gugus1234/security.py

Run Poetry once

poetry shell
poetry lock
poetry install

Admin User

let’s create admin user

How to Protect your App with Simple Security

Let’s build a small API Endpoint with FastAPI and protect it with SimpleSecurity.

API key based security package for FastAPI, focused on simplicity of use:

Full functionality out of the box, no configuration required
API key security with local sqlite backend, working with both header and query parameters
Default 15 days deprecation for generated API keys
Key creation, revocation, renewing, and usage logs handled through administrator endpoints
No dependencies, only requiring FastAPI and the python standard library

Build new App

and show the Directory Structure

FastAPI - Dependencies and Custom Headers

Source

https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/

Code

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

Test’s

Failed

no Custom Header

curl -s http://localhost/api/items/ |jq
{
  "detail": [
    {
      "loc": [
        "header",
        "x-token"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    },
    {
      "loc": [
        "header",
        "x-key"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

adding “x-token”

Fastapi

Fastapi - Ratelimiting