Rate Limiting

Overview

HFortix-Core provides two complementary rate-limiting facilities:

  • Tracking (RateLimitStats): visibility into API call patterns across multiple time windows. Informational only — it never blocks a request.

  • Enforcement (RateLimiter / AsyncRateLimiter): opt-in client-side throttling. Enabled by passing rate_limit=True (plus the rate_limit_* tuning kwargs) to any HTTP client or CloudSession constructor. Disabled by default.

Key Features (Tracking)

  • Multiple time windows: Tracks last minute, last 5 minutes, and last hour

  • Separate tracking: Tracks calls and errors independently

  • Configurable limits: Set custom limits or leave as None

  • Efficient implementation: Uses deque with maxlen for memory efficiency

  • Per-client tracking: Each service instance has its own RateLimitStats

  • Session-wide tracking: CloudSession tracks all calls across all services

  • No enforcement: RateLimitStats itself does not block requests — use the enforcement kwargs below for that

Time Windows

The system tracks statistics across three time windows:

  • Last minute (60 seconds): Short-term rate monitoring

  • Last 5 minutes (300 seconds): Medium-term trends

  • Last hour (3600 seconds): Long-term usage patterns

Both calls and errors are tracked separately for each window.

Basic Usage

FortiCare Example

from hfortix_forticare import FortiCare

# Configure custom limits (all optional, defaults to None)
fc = FortiCare(
    api_id="...",
    password="...",
    rate_limit_calls_per_min=100,      # 100 calls per minute
    rate_limit_calls_per_5min=500,     # 500 calls per 5 minutes
    rate_limit_calls_per_hour=1000,    # 1000 calls per hour
    rate_limit_errors_per_hour=10      # 10 errors per hour
)

# Make API calls
products = fc.api.products.list.post()

# Check rate limit status
status = fc.get_rate_limit_status()
print(f"Calls last min: {status['calls_last_min']}/{status['limits']['calls_per_min']}")
print(f"Calls last 5min: {status['calls_last_5min']}/{status['limits']['calls_per_5min']}")
print(f"Calls last hour: {status['calls_last_hour']}/{status['limits']['calls_per_hour']}")
print(f"Within limits: {status['within_limits']}")

FortiZTP Example

from hfortix_fortiztp import FortiZTP

# FortiZTP: 2000 calls per hour limit
client = FortiZTP(
    api_id="...",
    password="...",
    rate_limit_calls_per_hour=2000
)

# Check status
status = client.get_rate_limit_status()

Session-Wide Tracking

from hfortix_core.session import CloudSession
from hfortix_forticare import FortiCare
from hfortix_fortiztp import FortiZTP

with CloudSession(api_id="...", password="...") as session:
    fc = FortiCare(session=session)
    fz = FortiZTP(session=session)

    # Check per-client stats
    fc_status = fc.get_rate_limit_status()
    fz_status = fz.get_rate_limit_status()

    # Check session-wide stats (all services combined)
    session_status = session.get_rate_limit_status()
    print(f"Total calls (all services): {session_status['calls_last_hour']}")

Status Response Format

The get_rate_limit_status() method returns:

{
    "calls_last_min": 45,       # Calls in last 60 seconds
    "calls_last_5min": 180,     # Calls in last 300 seconds
    "calls_last_hour": 523,     # Calls in last 3600 seconds
    "errors_last_min": 0,       # Errors in last 60 seconds
    "errors_last_5min": 1,      # Errors in last 300 seconds
    "errors_last_hour": 2,      # Errors in last 3600 seconds
    "total_calls": 1247,        # Total since client creation
    "total_errors": 5,          # Total errors since creation
    "limits": {                 # Configured limits
        "calls_per_min": 100,
        "calls_per_5min": 500,
        "calls_per_hour": 1000,
        "errors_per_min": None,
        "errors_per_5min": None,
        "errors_per_hour": 10
    },
    "within_limits": True       # Whether within all set limits
}

Known API Rate Limits

FortiCare Asset Management

  • 100 calls per minute

  • 1000 calls per hour

  • 10 errors per hour

  • Batch operations: Max 10 units, max 5 errors per batch

FortiZTP

  • 2000 calls per hour

FortiGateCloud

  • 100 calls per second

  • 15 errors per minute

Enforcement

Enforcement is opt-in and implemented by RateLimiter (sync) and AsyncRateLimiter (async). All HTTP clients and CloudSession accept the same constructor kwargs:

from hfortix_core.http import HTTPClient

client = HTTPClient(
    url="https://192.168.1.99",
    token="...",
    rate_limit=True,                # Enable enforcement (default: False)
    rate_limit_max_requests=100,    # Allowed requests per window
    rate_limit_window_seconds=60.0, # Window length
    rate_limit_strategy="queue",    # "queue" | "drop" | "raise"
    rate_limit_queue_size=100,      # Max queued requests
    rate_limit_queue_timeout=30.0,  # Max seconds to wait in queue
    rate_limit_queue_overflow="block",  # "block" | "drop" | "raise"
)

Strategy behavior when the window limit is hit:

  • "queue" — requests wait for the next window (may raise RateLimitQueueFullError / RateLimitQueueTimeoutError depending on rate_limit_queue_overflow and rate_limit_queue_timeout)

  • "drop" — requests are silently dropped

  • "raise" — RateLimitExceededError is raised immediately

Note: the limiter is a per-window token counter, not a concurrency semaphore — RateLimiter.release() is intentionally a no-op; all accounting happens in acquire().

API Reference

RateLimitStats Class (tracking)

class hfortix_core.ratelimit.RateLimitStats(calls_per_min=None, calls_per_5min=None, calls_per_hour=None, errors_per_min=None, errors_per_5min=None, errors_per_hour=None)[source]

Bases: object

Simple rate limit tracking (no enforcement).

Tracks API calls and errors over multiple time windows: - Last minute (60 seconds) - Last 5 minutes (300 seconds) - Last hour (3600 seconds)

Does NOT enforce limits - just provides visibility.

Usage:
>>> stats = RateLimitStats(
...     calls_per_min=100,
...     calls_per_5min=500,
...     calls_per_hour=1000,
...     errors_per_hour=10
... )
>>>
>>> # Record calls
>>> stats.record_call()
>>> stats.record_error()
>>>
>>> # Check status
>>> status = stats.get_status()
>>> print(f"Calls last min: {status['calls_last_min']}")
>>> print(f"Calls last 5min: {status['calls_last_5min']}")
>>> print(f"Calls last hour: {status['calls_last_hour']}")
Parameters:

  • calls_per_min (int | None)

  • calls_per_5min (int | None)

  • calls_per_hour (int | None)

  • errors_per_min (int | None)

  • errors_per_5min (int | None)

  • errors_per_hour (int | None)

__init__(calls_per_min=None, calls_per_5min=None, calls_per_hour=None, errors_per_min=None, errors_per_5min=None, errors_per_hour=None)[source]

Initialize rate limit stats.

Parameters:

  • calls_per_min (int | None) – Maximum calls allowed per minute (None = no limit)

  • calls_per_5min (int | None) – Maximum calls allowed per 5 minutes (None = no limit)

  • calls_per_hour (int | None) – Maximum calls allowed per hour (None = no limit)

  • errors_per_min (int | None) – Maximum errors allowed per minute (None = no limit)

  • errors_per_5min (int | None) – Maximum errors allowed per 5 minutes (None = no limit)

  • errors_per_hour (int | None) – Maximum errors allowed per hour (None = no limit)

record_call()[source]

Record an API call.

Return type:

None

record_error()[source]

Record an API error.

Return type:

None

get_status()[source]

Get current rate limit status.

Returns:

  • calls_last_min: Call count in last 60 seconds

  • calls_last_5min: Call count in last 300 seconds

  • calls_last_hour: Call count in last 3600 seconds

  • errors_last_min: Error count in last 60 seconds

  • errors_last_5min: Error count in last 300 seconds

  • errors_last_hour: Error count in last 3600 seconds

  • total_calls: Total calls since creation

  • total_errors: Total errors since creation

  • limits: Configured limits

  • within_limits: True if all limits are respected (if set)

Return type:

Dict with

reset()[source]

Reset all counters (use with caution).

Return type:

None

RateLimiter Class (enforcement, sync)

AsyncRateLimiter Class (enforcement, async)

Implementation Details

Sliding Window Algorithm

The system uses collections.deque with maxlen to implement efficient sliding windows:

  1. Each call/error timestamp is appended to the deque

  2. When checking status, timestamps are filtered by time window

  3. Old timestamps beyond the window are automatically dropped

  4. Memory usage is bounded by maxlen

Thread Safety

  • RateLimitStats is not thread-safe by itself

  • CloudSession uses internal locking for thread-safe access

  • Per-client stats don’t require locking (single-threaded client usage assumed)

Future Enhancements

Planned features for future releases:

  • Adaptive throttling: Automatically slow down when approaching limits

  • Metrics export: Export statistics to Prometheus, CloudWatch, etc.

  • Historical tracking: Store and analyze rate limit trends over time

See Also