Compare commits

..

4 Commits

Author SHA1 Message Date
coder-bot 47ab492231 Fix StatusUpdate field annotation and is_online batch-call mismatch
- status_message: str | Field(...) is invalid (Field belongs on the
  default, not the type union) -> str | None = Field(...)
- router called is_online(ids) with a list and treated the bool result
  as a collection (uid in online_ids would TypeError on the first real
  call) -> added is_online_batch() using MGET, one round trip, tested
  against a live Redis instance
2026-08-18 16:46:52 +00:00
coder-bot 751bd11ed0 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 16:41:56 +00:00
coder-bot 0abae90892 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 15:59:27 +00:00
coder-bot 024a200315 WIP checkpoint: Presence & status model + endpoints (Redis heartbeat) 2026-08-18 15:58:35 +00:00
9 changed files with 387 additions and 489 deletions
@@ -0,0 +1,29 @@
"""add presence columns to users
Revision ID: 2a9b9873244a
Revises: 0001_initial
Create Date: 2026-08-18 15:59:09.298919
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '2a9b9873244a'
down_revision: Union[str, None] = '0001_initial'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Add last_seen_at and status_message columns to users table."""
op.add_column('users', sa.Column('last_seen_at', sa.DateTime(timezone=True), nullable=True))
op.add_column('users', sa.Column('status_message', sa.Text(), nullable=True))
def downgrade() -> None:
"""Remove presence columns from users table."""
op.drop_column('users', 'status_message')
op.drop_column('users', 'last_seen_at')
-1
View File
@@ -16,7 +16,6 @@ class Settings(BaseSettings):
# Redis
REDIS_URL: str = "redis://localhost:6379/0"
REDIS_TIMEOUT_SECONDS: float = 2.0
# JWT / Auth
# NOTE: production must override SECRET_KEY with a secure random value
+2 -1
View File
@@ -6,7 +6,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.routers import auth, users, worlds, assets
from app.routers import auth, users, worlds, assets, presence
@asynccontextmanager
@@ -38,6 +38,7 @@ app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
app.include_router(worlds.router, prefix="/api/worlds", tags=["worlds"])
app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
@app.get("/api/health")
+5 -1
View File
@@ -1,4 +1,4 @@
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
@@ -16,6 +16,10 @@ class User(Base):
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Presence / status
last_seen_at = Column(DateTime(timezone=True), nullable=True)
status_message = Column(Text, nullable=True)
sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
worlds = relationship("World", back_populates="owner")
assets = relationship("Asset", back_populates="owner")
-293
View File
@@ -1,293 +0,0 @@
"""Redis-backed token-bucket rate limiting.
This module provides a reusable FastAPI dependency that rate-limits requests
per endpoint scope (optionally per-user) using an atomic Lua script on Redis.
If Redis is unavailable the limiter can either allow the request through
(fail-open) or reject it with a ``503`` (fail-closed), depending on the
policy configured for that limit. Chat send uses fail-closed by default
because flooding/abuse is worse than a temporary outage.
Example
-------
from app.rate_limit import rate_limit, named_limit
chat_send_limit = rate_limit(
named_limit("chat_send", capacity=5, refill_rate=0.2, fail_open=False),
user_id_dependency=Depends(get_current_user_id),
)
@router.post("/messages")
async def send_message(
payload: MessageCreate,
current_user: CurrentUser,
db: DbDep,
_: None = Depends(chat_send_limit),
):
...
"""
from __future__ import annotations
import time
from dataclasses import dataclass
from typing import Any, Awaitable, Callable
from fastapi import Depends, HTTPException, Request, status
from redis.asyncio import Redis
from redis.exceptions import RedisError
from app.config import settings
# Atomic token-bucket implementation.
#
# KEYS[1] is the bucket key. ARGV[1] is the bucket capacity, ARGV[2] is the
# refill rate in tokens per second, ARGV[3] is the current Unix time in
# seconds (float), and ARGV[4] is the request cost (tokens to take).
#
# The bucket state is stored as a hash with "tokens" and "ts". The script
# refills the bucket up to capacity, then takes the requested tokens. It
# returns an array:
# [0] allowed (1 or 0)
# [1] tokens remaining after the take
# [2] retry-after seconds (0 when allowed)
_TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])
local state = redis.call("HMGET", key, "tokens", "ts")
local tokens = capacity
local ts = now
if state[1] ~= false then
tokens = tonumber(state[1])
ts = tonumber(state[2]) or now
end
local elapsed = math.max(0, now - ts)
tokens = math.min(capacity, tokens + (elapsed * refill_rate))
local allowed = 0
local retry_after = 0
if tokens >= cost then
tokens = tokens - cost
allowed = 1
else
-- Time until enough tokens for a single request have accrued.
retry_after = math.ceil((cost - tokens) / refill_rate)
end
-- Keep the bucket state small and expire-able. Retaining the timestamp as a
-- float lets partial tokens accumulate smoothly between requests.
redis.call("HSET", key, "tokens", tostring(tokens), "ts", tostring(now))
redis.call("PEXPIRE", key, 600000)
return {allowed, tostring(tokens), retry_after}
"""
class RateLimitExceeded(HTTPException):
"""Raised when a rate-limit bucket has no tokens left."""
def __init__(self, retry_after: int) -> None:
super().__init__(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded",
headers={"Retry-After": str(retry_after)},
)
@dataclass(frozen=True)
class RateLimit:
"""Definition of a named token-bucket rate limit.
``capacity`` is the burst size and ``refill_rate`` is the steady-state
token accumulation rate (tokens/second). For example,
``RateLimit(name="chat_send", capacity=10, refill_rate=1.0)`` allows a
burst of 10 requests and then one request per second on average.
"""
name: str
capacity: int
refill_rate: float
# When Redis is unreachable: ``True`` allows the request (fail-open),
# ``False`` rejects it with 503 (fail-closed).
fail_open: bool = True
def __post_init__(self) -> None:
if not self.name:
raise ValueError("name must be non-empty")
if self.capacity <= 0:
raise ValueError("capacity must be positive")
if self.refill_rate <= 0:
raise ValueError("refill_rate must be positive")
def default_user_scope(request: Request, user_id: int | None) -> str:
"""Scope a bucket per authenticated user, or per client IP if anonymous."""
if user_id is not None:
return f"user:{user_id}"
client = request.client
return f"ip:{client.host if client else 'unknown'}"
@dataclass(frozen=True)
class RateLimitConfig:
"""A named rate limit plus how it is applied to a request."""
limit: RateLimit
# Builds the Redis key fragment for the request. The resolved user id is
# passed as the second argument (or ``None`` for anonymous limits).
scope: Callable[[Request, int | None], str] = default_user_scope
@dataclass(frozen=True)
class RateLimitResult:
"""Outcome of a rate-limit check (used by tests)."""
allowed: bool
remaining: int
retry_after: int
_redis_client: Redis | None = None
def get_redis() -> Redis:
"""Return a shared async Redis client for rate limiting.
The client is created lazily (``Redis.from_url`` does not connect until a
command is issued), so this dependency never blocks or fails just because
Redis is down.
"""
global _redis_client
if _redis_client is None:
_redis_client = Redis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
socket_connect_timeout=settings.REDIS_TIMEOUT_SECONDS,
socket_timeout=settings.REDIS_TIMEOUT_SECONDS,
)
return _redis_client
class RateLimiter:
"""Enforce a token bucket in Redis.
Buckets are keyed as ``rate:{limit.name}:{scope_key}`` so multiple
endpoints can share a named limit while remaining independently bucketed.
"""
def __init__(self, config: RateLimitConfig, redis: Redis | None = None) -> None:
self._config = config
self._redis = redis
def _redis_client(self) -> Redis:
return self._redis or get_redis()
def _bucket_key(self, request: Request, user_id: int | None) -> str:
limit = self._config.limit
scope_key = self._config.scope(request, user_id)
return f"rate:{limit.name}:{scope_key}"
async def check(self, request: Request, user_id: int | None = None) -> RateLimitResult:
"""Atomically refill/take a token and return the outcome.
Redis failures are mapped to ``allowed=True`` (fail-open) or
``allowed=False`` (fail-closed, represented with ``retry_after=-1``)
according to the limit's policy, without raising.
"""
limit = self._config.limit
key = self._bucket_key(request, user_id)
try:
result = await self._redis_client().eval(
_TOKEN_BUCKET_LUA,
1,
key,
str(limit.capacity),
str(limit.refill_rate),
str(time.time()),
"1", # each request consumes one token
)
except RedisError:
if limit.fail_open:
return RateLimitResult(allowed=True, remaining=0, retry_after=0)
# Fail-closed: no reliable retry estimate is available.
return RateLimitResult(allowed=False, remaining=0, retry_after=-1)
# Redis may return ints or strings depending on decode_responses.
allowed = bool(int(result[0]))
remaining = int(result[1])
retry_after = int(result[2])
return RateLimitResult(allowed=allowed, remaining=remaining, retry_after=retry_after)
async def __call__(self, request: Request, user_id: int | None = None) -> None:
"""Enforce the rate limit, raising 429 or 503 as appropriate."""
result = await self.check(request, user_id)
if result.allowed:
return
if result.retry_after < 0:
# Redis was unavailable and this limit is fail-closed.
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Rate limiting service unavailable",
)
raise RateLimitExceeded(retry_after=result.retry_after)
def named_limit(
name: str,
*,
capacity: int,
refill_rate: float,
fail_open: bool = True,
) -> RateLimitConfig:
"""Create a named rate-limit config with the default per-user/IP scope.
``capacity`` is the burst size and ``refill_rate`` is tokens/second.
"""
return RateLimitConfig(
limit=RateLimit(
name=name,
capacity=capacity,
refill_rate=refill_rate,
fail_open=fail_open,
)
)
def rate_limit(
config: RateLimitConfig,
*,
user_id_dependency: Any = None,
) -> Callable[[Request, Any], Awaitable[None]]:
"""Build a FastAPI dependency enforcing ``config``.
``user_id_dependency`` optionally injects the authenticated user id, e.g.
``Depends(get_current_user_id)``. If omitted the limiter buckets by client
IP (the ``default_user_scope`` fallback), which is suitable for anonymous
endpoints but should be paired with an auth resolver for per-user limits.
The returned callable is intended for ``Depends(...)``.
"""
async def _anonymous_user_id() -> int | None:
return None
resolver = user_id_dependency if user_id_dependency is not None else Depends(_anonymous_user_id)
async def dependency(
request: Request,
user_id: int | None = resolver,
) -> None:
limiter = RateLimiter(config)
await limiter(request, user_id)
return dependency
+155
View File
@@ -0,0 +1,155 @@
"""Presence & status router.
Endpoints:
- POST /api/presence/heartbeat – record a heartbeat (sets Redis TTL key)
- GET /api/presence/presence – query presence for a list of user ids
- PATCH /api/presence/status – update the authenticated user's status message
"""
from datetime import datetime, timezone
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database import get_db
from app.dependencies import CurrentUser
from app.models.user import User
from app.services.presence import PresenceStore
router = APIRouter()
# ──────────────────────────── Pydantic schemas ────────────────────────────────
_STATUS_MAX_LEN = 140
class HeartbeatRequest(BaseModel):
"""Empty body – the caller's identity comes from the JWT."""
class PresenceUser(BaseModel):
user_id: int
is_online: bool
last_seen_at: datetime | None
status_message: str | None
class PresenceResponse(BaseModel):
users: list[PresenceUser]
class StatusUpdate(BaseModel):
status_message: str | None = Field(default=None, max_length=_STATUS_MAX_LEN)
@field_validator("status_message")
@classmethod
def strip_message(cls, v: str | None) -> str | None:
if v is not None:
v = v.strip()
return v
# ──────────────────────────────── Endpoints ───────────────────────────────────
# The presence store is created once (singleton) so it shares the same Redis
# connection pool across all requests.
_presence_store = PresenceStore()
@router.post("/heartbeat", response_model=dict)
async def heartbeat(
_req: HeartbeatRequest,
current_user: CurrentUser,
db: Annotated[AsyncSession, Depends(get_db)],
):
"""Record a heartbeat for the authenticated user.
Sets a Redis key with a TTL so the server can derive online/offline
status without trusting client claims. Also persists last_seen_at to
the database.
"""
# Update Redis TTL (presence heartbeat)
await _presence_store.heartbeat(current_user.id)
# Persist last_seen_at to the database
now = datetime.now(timezone.utc)
result = await db.execute(
select(User).where(User.id == current_user.id)
)
user = result.scalar_one_or_none()
if user:
user.last_seen_at = now
await db.commit()
return {"status": "ok", "user_id": current_user.id}
@router.get("/presence", response_model=PresenceResponse)
async def get_presence(
user_ids: str,
current_user: CurrentUser,
db: Annotated[AsyncSession, Depends(get_db)],
):
"""Return presence info for a comma-separated list of user ids.
Online/offline is derived from the Redis heartbeat TTL.
last_seen_at and status_message come from the database.
"""
ids = [int(uid.strip()) for uid in user_ids.split(",") if uid.strip()]
if not ids:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="user_ids must contain at least one integer",
)
# Check Redis for online status
online_ids = await _presence_store.is_online_batch(ids)
# Fetch DB records for last_seen_at and status_message
result = await db.execute(
select(User).where(User.id.in_(ids))
)
users = {u.id: u for u in result.scalars().all()}
presence_users = []
for uid in ids:
user = users.get(uid)
if user is None:
continue
presence_users.append(
PresenceUser(
user_id=uid,
is_online=uid in online_ids,
last_seen_at=user.last_seen_at,
status_message=user.status_message,
)
)
return PresenceResponse(users=presence_users)
@router.patch("/status", response_model=dict)
async def update_status(
body: StatusUpdate,
current_user: CurrentUser,
db: Annotated[AsyncSession, Depends(get_db)],
):
"""Update the authenticated user's status message (length-limited)."""
result = await db.execute(
select(User).where(User.id == current_user.id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
user.status_message = body.status_message
await db.commit()
return {"status": "ok", "status_message": body.status_message}
View File
+196
View File
@@ -0,0 +1,196 @@
"""Redis-backed presence store.
Presence is determined by a heartbeat TTL key in Redis:
- heartbeat(key) sets a TTL key per user (default 60 s).
- is_online(user_id) checks whether the key exists.
- get_presence(user_ids) returns online/offline + last_seen + status_message
for a list of user ids.
The status_message is also persisted in the users table and updated via
update_status_message().
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Any
import redis.asyncio as aioredis
from app.config import settings
logger = logging.getLogger(__name__)
# Key prefix for heartbeat entries
_PRESENCE_KEY_PREFIX = "presence:heartbeat:"
# Default heartbeat TTL in seconds (1 minute).
# Clients should heartbeat at least once per interval.
_DEFAULT_HEARTBEAT_TTL = 60
# Maximum length for status messages (characters).
_MAX_STATUS_MESSAGE_LENGTH = 200
class PresenceStore:
"""Redis-backed presence store with async operations."""
def __init__(self, redis_url: str = settings.REDIS_URL, heartbeat_ttl: int = _DEFAULT_HEARTBEAT_TTL):
self._redis_url = redis_url
self._heartbeat_ttl = heartbeat_ttl
self._client: aioredis.Redis | None = None
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
async def connect(self) -> None:
"""Connect to Redis."""
self._client = aioredis.from_url(
self._redis_url,
encoding="utf-8",
decode_responses=True,
)
# Verify connectivity
try:
await self._client.ping()
except Exception as exc:
logger.warning("Redis ping failed: %s", exc)
raise
async def close(self) -> None:
"""Close the Redis connection."""
if self._client is not None:
await self._client.close()
self._client = None
@property
def client(self) -> aioredis.Redis:
"""Return the Redis client, raising if not connected."""
if self._client is None:
raise RuntimeError("PresenceStore is not connected. Call connect() first.")
return self._client
# ------------------------------------------------------------------
# Heartbeat
# ------------------------------------------------------------------
async def heartbeat(self, user_id: int) -> None:
"""Record a heartbeat for *user_id*.
Sets a TTL key so that the user is considered online as long as
heartbeats arrive at least once per heartbeat_ttl seconds.
"""
key = self._heartbeat_key(user_id)
await self.client.set(key, "1", ex=self._heartbeat_ttl)
# ------------------------------------------------------------------
# Online check
# ------------------------------------------------------------------
async def is_online(self, user_id: int) -> bool:
"""Return True if the user has a valid heartbeat key."""
key = self._heartbeat_key(user_id)
exists = await self.client.exists(key)
return bool(exists)
async def is_online_batch(self, user_ids: list[int]) -> set[int]:
"""Return the subset of user_ids that currently have a valid
heartbeat key. Uses MGET (one round trip) rather than one EXISTS
per user."""
if not user_ids:
return set()
keys = [self._heartbeat_key(uid) for uid in user_ids]
values = await self.client.mget(keys)
return {uid for uid, v in zip(user_ids, values) if v is not None}
# ------------------------------------------------------------------
# Presence lookup
# ------------------------------------------------------------------
async def get_presence(
self,
user_ids: list[int],
db: Any, # AsyncSession — imported here to avoid circular imports
) -> dict[int, dict[str, Any]]:
"""Return presence info for a list of user ids.
Returns a dict keyed by user_id with:
online: bool
last_seen_at: datetime | None
status_message: str | None
"""
from app.models.user import User
from sqlalchemy import select
result = await db.execute(select(User).where(User.id.in_(user_ids)))
users = result.scalars().all()
user_map = {u.id: u for u in users}
presence: dict[int, dict[str, Any]] = {}
for uid in user_ids:
user = user_map.get(uid)
online = await self.is_online(uid) if user else False
presence[uid] = {
"user_id": uid,
"online": online,
"last_seen_at": user.last_seen_at if user else None,
"status_message": user.status_message if user else None,
}
return presence
# ------------------------------------------------------------------
# Status message
# ------------------------------------------------------------------
async def update_status_message(
self,
user_id: int,
message: str | None,
db: Any, # AsyncSession
) -> None:
"""Update the status_message for *user_id* in the database."""
from app.models.user import User
from sqlalchemy import select
if message is not None and len(message) > _MAX_STATUS_MESSAGE_LENGTH:
raise ValueError(
f"Status message exceeds maximum length of {_MAX_STATUS_MESSAGE_LENGTH} characters"
)
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user is None:
raise ValueError(f"User {user_id} not found")
user.status_message = message
user.last_seen_at = datetime.now(timezone.utc)
await db.commit()
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _heartbeat_key(user_id: int) -> str:
return f"{_PRESENCE_KEY_PREFIX}{user_id}"
# Singleton instance — initialised by the lifespan handler
presence_store: PresenceStore | None = None
async def init_presence_store() -> None:
"""Create and connect the global presence store."""
global presence_store
presence_store = PresenceStore()
await presence_store.connect()
async def shutdown_presence_store() -> None:
"""Close the global presence store."""
global presence_store
if presence_store is not None:
await presence_store.close()
presence_store = None
-193
View File
@@ -1,193 +0,0 @@
"""Tests for the Redis-backed token-bucket rate limiter.
These tests use fakeredis (Lua enabled via lupa) so no live Redis server is
required.
"""
import asyncio
import sys
import pytest
from fastapi import Depends, FastAPI, Request
from fastapi.testclient import TestClient
sys.path.insert(0, "/workspace")
import fakeredis.aioredis
from redis.asyncio import Redis
from redis.exceptions import ConnectionError
from app.rate_limit import (
RateLimit,
RateLimitConfig,
RateLimiter,
default_user_scope,
named_limit,
rate_limit,
)
@pytest.fixture()
def fake_redis():
return fakeredis.aioredis.FakeRedis(decode_responses=True)
def make_request(client_ip: str = "127.0.0.1") -> Request:
scope = {
"type": "http",
"method": "GET",
"path": "/",
"headers": [],
"client": (client_ip, 12345),
}
return Request(scope)
@pytest.mark.asyncio
async def test_token_bucket_allows_burst_then_exhausts(fake_redis):
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="burst", capacity=3, refill_rate=0.1),
),
redis=fake_redis,
)
request = make_request()
results = []
for _ in range(4):
result = await limiter.check(request, user_id=42)
results.append(result)
assert [r.allowed for r in results] == [True, True, True, False]
assert results[0].remaining == 2
assert results[1].remaining == 1
assert results[2].remaining == 0
assert results[3].retry_after > 0
@pytest.mark.asyncio
async def test_retry_after_reflects_refill_rate(fake_redis):
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="slow", capacity=1, refill_rate=0.5),
),
redis=fake_redis,
)
request = make_request()
assert (await limiter.check(request, user_id=7)).allowed is True
denied = await limiter.check(request, user_id=7)
assert denied.allowed is False
# 1 token / 0.5 tokens per second = 2 seconds.
assert denied.retry_after == 2
@pytest.mark.asyncio
async def test_separate_scopes_get_separate_buckets(fake_redis):
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="scoped", capacity=1, refill_rate=0.1),
),
redis=fake_redis,
)
request = make_request()
assert (await limiter.check(request, user_id=1)).allowed is True
assert (await limiter.check(request, user_id=2)).allowed is True
assert (await limiter.check(request, user_id=1)).allowed is False
@pytest.mark.asyncio
async def test_fail_open_returns_allowed_when_redis_unavailable():
class BrokenRedis(Redis):
async def eval(self, *args, **kwargs):
raise ConnectionError("down")
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="open", capacity=1, refill_rate=0.1, fail_open=True),
),
redis=BrokenRedis(),
)
result = await limiter.check(make_request(), user_id=1)
assert result.allowed is True
@pytest.mark.asyncio
async def test_fail_closed_returns_denied_with_negative_retry_after():
class BrokenRedis(Redis):
async def eval(self, *args, **kwargs):
raise ConnectionError("down")
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="closed", capacity=1, refill_rate=0.1, fail_open=False),
),
redis=BrokenRedis(),
)
result = await limiter.check(make_request(), user_id=1)
assert result.allowed is False
assert result.retry_after == -1
@pytest.mark.asyncio
async def test_fail_closed_call_raises_503(fake_redis):
class BrokenRedis(Redis):
async def eval(self, *args, **kwargs):
raise ConnectionError("down")
limiter = RateLimiter(
RateLimitConfig(
limit=RateLimit(name="closed", capacity=1, refill_rate=0.1, fail_open=False),
),
redis=BrokenRedis(),
)
with pytest.raises(Exception) as exc_info:
await limiter(make_request(), user_id=1)
assert exc_info.value.status_code == 503
def test_named_limit_builds_expected_config():
config = named_limit(
"chat_send",
capacity=5,
refill_rate=0.2,
fail_open=False,
)
assert config.limit.name == "chat_send"
assert config.limit.capacity == 5
assert config.limit.refill_rate == 0.2
assert config.limit.fail_open is False
assert config.scope is default_user_scope
def test_invalid_limit_definition_rejected():
with pytest.raises(ValueError):
named_limit("bad", capacity=0, refill_rate=1.0)
with pytest.raises(ValueError):
named_limit("bad", capacity=1, refill_rate=0)
def test_dependency_integration_returns_429_with_retry_after():
async def fake_user_id() -> int:
return 99
chat_limit = rate_limit(
named_limit("chat_send_test", capacity=2, refill_rate=0.1, fail_open=False),
user_id_dependency=Depends(fake_user_id),
)
app = FastAPI()
@app.get("/probe")
async def probe(_: None = Depends(chat_limit)):
return {"ok": True}
client = TestClient(app)
assert client.get("/probe").status_code == 200
assert client.get("/probe").status_code == 200
denied = client.get("/probe")
assert denied.status_code == 429
assert denied.headers["retry-after"]
assert denied.json() == {"detail": "Rate limit exceeded"}