CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://127.0.0.1:6379/1",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
},
}
}
REST_FRAMEWORK = {
"DEFAULT_THROTTLE_RATES": {
"public": "60/min",
},
"EXCEPTION_HANDLER": "rest_framework.views.exception_handler",
}
import time
import uuid
from django.core.cache import cache
from rest_framework.throttling import BaseThrottle
class SlidingWindowThrottle(BaseThrottle):
scope = "public"
default_rate = "60/min"
def __init__(self):
self.num_requests, self.duration = self.parse_rate(self.default_rate)
def parse_rate(self, rate):
num, period = rate.split("/")
seconds = {"s": 1, "min": 60, "hour": 3600, "day": 86400}[period]
return int(num), seconds
def get_cache_key(self, request, view):
if request.user and request.user.is_authenticated:
ident = f"user:{request.user.pk}"
else:
ident = f"ip:{self.get_ident(request)}"
return f"throttle:{self.scope}:{ident}"
def allow_request(self, request, view):
self.key = self.get_cache_key(request, view)
now = time.time()
window_start = now - self.duration
client = cache.client.get_client(write=True)
member = f"{now}:{uuid.uuid4().hex}"
pipe = client.pipeline()
pipe.zremrangebyscore(self.key, 0, window_start)
pipe.zadd(self.key, {member: now})
pipe.zcard(self.key)
pipe.expire(self.key, self.duration + 1)
_, _, count, _ = pipe.execute()
if count > self.num_requests:
client.zrem(self.key, member) # do not charge rejected calls
self.now = now
return False
return True
def wait(self):
client = cache.client.get_client(write=False)
oldest = client.zrange(self.key, 0, 0, withscores=True)
if not oldest:
return None
_, oldest_ts = oldest[0]
return max(0.0, (oldest_ts + self.duration) - self.now)
from rest_framework.generics import ListAPIView
from rest_framework.permissions import AllowAny
from .models import Quote
from .serializers import QuoteSerializer
from .throttling import SlidingWindowThrottle
class PublicQuoteView(ListAPIView):
permission_classes = [AllowAny]
throttle_classes = [SlidingWindowThrottle]
serializer_class = QuoteSerializer
queryset = Quote.objects.filter(is_public=True).order_by("-created_at")
def get_throttles(self):
throttles = super().get_throttles()
if self.request.query_params.get("preview"):
# tighter limit for the unauthenticated preview path
for t in throttles:
t.num_requests = 10
return throttles
This snippet shows how a public Django REST Framework endpoint is protected from abuse with a custom sliding-window throttle backed by Redis. Django's built-in SimpleRateThrottle uses a fixed-window algorithm that stores a list of timestamps per key in the cache; the problem is that fixed windows allow bursts at window boundaries (a client can fire the full quota at the end of one window and again at the start of the next). The SlidingWindowThrottle in throttling.py fixes that by keeping every request timestamp in a sorted set and counting only the timestamps inside a rolling duration, giving a smoother, harder-to-game limit.
In throttling.py, SlidingWindowThrottle subclasses DRF's BaseThrottle rather than SimpleRateThrottle because the counting logic is bespoke. The get_cache_key method identifies the caller: authenticated users key on pk, anonymous callers key on the client IP resolved via get_ident, so a shared limit is applied per-IP for the public case. allow_request executes a small Redis pipeline: it removes timestamps older than the window with zremrangebyscore, adds the current request with a unique member, reads the cardinality with zcard, and refreshes the key TTL with expire. Doing all four in one pipeline makes the read-modify-write effectively atomic, which matters under concurrent traffic where two requests could otherwise both pass a stale count.
When the count exceeds num_requests, the pre-inserted member is rolled back with zrem so a rejected request does not consume a slot, and allow_request returns False. The wait method computes how long until the oldest in-window request expires, which DRF surfaces as the Retry-After header.
In views.py, PublicQuoteView simply lists the throttle in throttle_classes; DRF calls allow_request before the handler runs and raises Throttled automatically on rejection. settings.py wires up the Redis-backed cache and a default rate string parsed by parse_rate. The main trade-off is memory and Redis round-trips: a sorted set per client is heavier than a single counter, so the sliding window suits endpoints where fairness matters more than raw throughput. Keeping the TTL slightly longer than duration ensures idle keys are reaped without dropping timestamps that are still relevant.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
module Api
module V1
class UsersController < BaseController
def show
user = User.includes(:profile).find(params[:id])ETags for conditional requests and caching
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
type User {
id: ID!
name: String!
email: String!
posts: [Post!]!
createdAt: String!GraphQL API with Spring Boot
Share this code
Here's the card — post it anywhere.
