from django.core.cache import cache
class RateLimiter:
def __init__(self, scope, limit, window_seconds):
self.scope = scope
self.limit = limit
self.window = window_seconds
def _key(self, identifier):
return "ratelimit:{}:{}".format(self.scope, identifier)
def hit(self, identifier):
key = self._key(identifier)
cache.add(key, 0, self.window)
try:
return cache.incr(key)
except ValueError:
# Key expired between add() and incr(); reseed and count this hit.
cache.add(key, 1, self.window)
return 1
def is_exceeded(self, identifier):
return self.hit(identifier) > self.limit
def remaining(self, identifier):
current = cache.get(self._key(identifier), 0)
return max(0, self.limit - current)
from django.contrib.auth.forms import PasswordResetForm
from django.core.exceptions import ValidationError
from .throttle import RateLimiter
EMAIL_LIMITER = RateLimiter("pwreset:email", limit=3, window_seconds=3600)
IP_LIMITER = RateLimiter("pwreset:ip", limit=10, window_seconds=3600)
class ThrottledPasswordResetForm(PasswordResetForm):
def __init__(self, *args, **kwargs):
self.request = kwargs.pop("request", None)
super().__init__(*args, **kwargs)
def clean(self):
cleaned = super().clean()
email = cleaned.get("email")
if not email:
return cleaned
ip = getattr(self.request, "client_ip", None) or "unknown"
if IP_LIMITER.is_exceeded(ip):
raise ValidationError(
"Too many password reset requests from your network. "
"Please try again later."
)
if EMAIL_LIMITER.is_exceeded(email.lower()):
raise ValidationError(
"This address has requested too many resets recently. "
"Please wait before trying again."
)
return cleaned
from django.contrib.auth.views import PasswordResetView
from django.urls import reverse_lazy
from .forms import ThrottledPasswordResetForm
class PasswordResetRequestView(PasswordResetView):
form_class = ThrottledPasswordResetForm
template_name = "registration/password_reset_form.html"
email_template_name = "registration/password_reset_email.html"
success_url = reverse_lazy("password_reset_done")
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
self.request.client_ip = self._client_ip()
kwargs["request"] = self.request
return kwargs
def _client_ip(self):
forwarded = self.request.META.get("HTTP_X_FORWARDED_FOR")
if forwarded:
# Left-most entry is the original client behind trusted proxies.
return forwarded.split(",")[0].strip()
return self.request.META.get("REMOTE_ADDR", "unknown")
This snippet shows a practical way to throttle password reset attempts in Django by combining a small cache helper, a custom form, and a view that consumes them. The goal is to stop an abuser (or a runaway script) from hammering the reset endpoint to enumerate accounts or flood users with reset emails, without adding a heavyweight rate-limiting dependency.
In throttle.py, RateLimiter wraps Django's cache framework to implement a fixed-window counter. hit() builds a scoped key from a scope and identifier, then uses cache.add() to seed the counter to zero only when the key is absent — add() is atomic and sets the TTL, so the window's expiry is fixed at first contact and won't be extended by later hits. cache.incr() then bumps the count; because there is a race between add and incr, the code guards against a ValueError when the key expired mid-flight and retries the seed. remaining() reports how many attempts are left, useful for surfacing a friendly message.
In forms.py, ThrottledPasswordResetForm subclasses Django's built-in PasswordResetForm and enforces the limit inside clean(). It rate-limits on two axes: the submitted email and the client IP, so neither a single address nor a single host can dominate. Crucially, throttling happens in the form rather than the view because clean() is the natural place for cross-field validation and keeps the rule reusable across any view that instantiates the form. When the limit is exceeded, it raises ValidationError, which Django renders as a normal form error.
The form needs the request to read the IP, so PasswordResetRequestView in views.py overrides get_form_kwargs() to pass request through, and _client_ip() prefers X-Forwarded-For when behind a proxy.
The main trade-off is that a fixed window allows bursts at window boundaries, unlike a sliding window or token bucket. It is also only as strong as the shared cache backend — LocMemCache per-process defeats it, so a real deployment should back this with Redis or Memcached. Trusting X-Forwarded-For blindly is a pitfall; only honor it behind a proxy that sets it.
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
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
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
json.array! @posts do |post|
json.cache! ['v1', post], expires_in: 1.hour do
json.id post.id
json.title post.title
json.excerpt post.excerpt
json.published_at post.published_atFragment caching for expensive JSON serialization
Share this code
Here's the card — post it anywhere.
