active_request

active_request is a tiny Redis-backed sliding-window counter for tracking live requests in the last N seconds (default 60s). It is safe across multiple processes and works with both sync and async Redis clients (e.g. FastAPI).

Installation

pip install active_request

Requires Redis server and redis Python client ≥ 5.0.

Usage

Sync

from __future__ import annotations

import os

import redis
from active_request import record_active_request

r = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))

count = record_active_request(
    r,
    key="metrics:active_requests:api",
    window_seconds=60,
    gauge_key="metrics:active_requests:api:count",
)
print(count)  # live number of requests in the last 60s

Async (FastAPI example)

from __future__ import annotations

import os
from fastapi import FastAPI, Depends
from redis.asyncio import Redis

from active_request import arecord_active_request

app = FastAPI()
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
r = Redis.from_url(redis_url)


@app.get("/ping")
async def ping():
    # Increment and get live count of requests in the last 60 seconds
    live = await arecord_active_request(
        r,
        key="metrics:active_requests:api",
        window_seconds=60,
        gauge_key="metrics:active_requests:api:count",
    )
    return {"ok": True, "live_60s": live}

Features

• Sliding window counter using a Redis sorted set + Lua (accurate to milliseconds).
• Concurrency-safe across multiple workers/processes.
• Sync and async APIs: record_active_request / arecord_active_request.
• Optional “gauge” key written with short TTL for easy scraping/dashboards.
• Small dependency footprint (redis only).

Contributing

Contributions, issues, and feature requests are welcome! Repo: https://github.com/chigwell/active_request • Issues: https://github.com/chigwell/active_request/issues

License

active_request is licensed under the MIT License.