domain-api-limiter 0.2.0

Rate-limit declaration toolkit: throttle policies, tier rates, and declaration decorators for Python services

domain-api-limiter

Declarative rate-limit policy toolkit for Python services. Attach throttle policies to methods, parse rates, define tier overrides, and collect policies for consumption by framework-level throttling adapters.

Why domain-api-limiter?

Rate-limiting policies are declarations: they define what throttle rules apply to a service method, but they never enforce the limit themselves. Enforcement belongs in the consuming framework's vetted throttling logic (Django REST Framework throttles, Starlette middleware, etc.). domain-api-limiter keeps the declaration layer simple, immutable, and composable. Policies are frozen dataclasses, validated at import time so malformed rates fail fast. The @throttled decorator attaches policy metadata to any method without changing its signature. PolicyRegistry collects declared policies from a service class or module, returning them in definition order so an adapter can build the framework's real throttle classes. Tier overrides let you apply different rate limits by customer plan (free/pro/enterprise) without duplicating policy declarations.

Installation

pip install domain-api-limiter

Or with uv:

uv add domain-api-limiter

Requires Python 3.11+. Depends on domain-errors >= 0.1.0.

Quick Start

1. Declare throttle policies with @throttled (method or class level)

Apply @throttled to individual methods:

from domain_api_limiter import throttled

class DocumentService:
    @throttled("docs:list", "100/hour", tiers={"free": "10/day", "pro": "1000/hour"})
    def list_documents(self, tenant_id: str) -> list:
        """List all documents for a tenant."""
        return []

    @throttled("docs:upload", "20/minute")
    def upload(self, tenant_id: str) -> None:
        """Upload a document."""
        pass

Or apply to the entire class; each public method gets scope root:method_name:

@throttled("docs", "100/hour", tiers={"free": "10/day", "pro": "1000/hour"})
class DocumentService:
    def list_documents(self, tenant_id: str) -> list:
        """List all documents for a tenant."""
        return []

    def upload(self, tenant_id: str) -> None:
        """Upload a document."""
        pass

    def _internal_helper(self) -> None:
        """Private methods are skipped."""
        pass

In class-level mode, list_documents gets scope docs:list_documents and upload gets docs:upload. Private methods (_-prefixed), dunders, properties, and nested classes are skipped. Methods already decorated with @throttled keep their individual policies.

The decorator validates the rate and tier strings at decoration time and attaches a ThrottlePolicy without changing method behavior.

Class-level @throttled with PolicyRegistry.collect() output:

from domain_api_limiter import throttled, PolicyRegistry

@throttled("docs", "100/hour", tiers={"free": "10/day", "pro": "1000/hour"})
class DocumentService:
    def list_documents(self, tenant_id: str) -> list:
        return []

    def upload(self, tenant_id: str) -> None:
        pass

    @throttled("docs:delete", "50/day")
    def delete_document(self, doc_id: str) -> None:
        """Overrides class-level policy."""
        pass

    def _internal_helper(self) -> None:
        """Private methods are skipped."""
        pass

    @staticmethod
    def validate_name(name: str) -> bool:
        return bool(name)

registry = PolicyRegistry()
policies = registry.collect(DocumentService)

Output:

Collected 4 policies from DocumentService:

Scope: docs:list_documents
Base rate: 100/hour
  free: 10/day
  pro: 1000/hour

Scope: docs:upload
Base rate: 100/hour
  free: 10/day
  pro: 1000/hour

Scope: docs:delete
Base rate: 50/day

Scope: docs:validate_name
Base rate: 100/hour
  free: 10/day
  pro: 1000/hour

2. Collect policies for your adapter

from domain_api_limiter import PolicyRegistry

registry = PolicyRegistry()
policies = registry.collect(DocumentService)
print(f"Collected {len(policies)} policies:")
# Output: Collected 2 policies:
#   docs:list: 100/hour (tiers: free=10/day, pro=1000/hour)
#   docs:upload: 20/minute

Your adapter then walks the policies and builds the framework's real throttle classes (DRF throttles, middleware, etc.).

3. Parse rates and inspect tier overrides

from domain_api_limiter import RateLimit, Period

# Parse rate strings
rate = RateLimit.from_rate("100/hour")
print(rate.requests)      # 100
print(rate.period)         # Period.HOUR
print(rate.period_seconds) # 3600
print(rate.as_rate())      # "100/hour"

# Create rates directly
rate = RateLimit(requests=50, period=Period.MINUTE)

# Inspect tier overrides on a collected policy
policy = policies[0]
if policy.has_tiers:
    free_rate = policy.rate_for("free")       # Returns RateLimit for "free" tier
    pro_rate = policy.rate_for("pro")         # Returns RateLimit for "pro" tier
    default_rate = policy.rate_for("unknown") # Returns the base rate

4. Handle declaration errors

from domain_api_limiter import ThrottleDeclarationError

# Errors are raised at decoration time, not at runtime.
try:
    @throttled("scope", "invalid")  # Malformed rate
    def bad_method():
        pass
except ThrottleDeclarationError as e:
    print(f"Invalid policy: {e.message}")

Service Class Example

A typical service class with throttled methods:

from domain_api_limiter import throttled, PolicyRegistry

class DocumentService:
    @throttled("docs:list", "100/hour", tiers={"free": "10/day", "pro": "1000/hour"})
    def list_documents(self, tenant_id: str) -> list:
        """List all documents for a tenant."""
        return []

    @throttled("docs:upload", "20/minute")
    def upload(self, tenant_id: str) -> None:
        """Upload a document."""
        pass

# An adapter collects policies and builds throttle classes
registry = PolicyRegistry()
policies = registry.collect(DocumentService)

for policy in policies:
    print(f"Scope: {policy.scope}")
    print(f"Base rate: {policy.rate.as_rate()}")
    if policy.has_tiers:
        for tier_rate in policy.tier_rates:
            print(f"  {tier_rate.tier}: {tier_rate.rate.as_rate()}")

Output:

Scope: docs:list
Base rate: 100/hour
  free: 10/day
  pro: 1000/hour
Scope: docs:upload
Base rate: 20/minute

Public API

Throttle Declaration

from domain_api_limiter import throttled, ThrottlePolicy, RateLimit, TierRate, Period

# Decorator: attach a policy to a method
@throttled(scope: str, rate: str, tiers: dict[str, str] | None = None)
def my_method() -> None:
    pass

# Value objects (immutable, validated at creation)
policy = ThrottlePolicy(
    scope: str,                      # Policy identifier (e.g., "docs:list")
    rate: RateLimit,                 # Base rate for all tiers
    tier_rates: tuple[TierRate, ...] # Per-tier rate overrides (default: ())
)

rate = RateLimit(
    requests: int,  # Positive count of allowed requests
    period: Period  # Period enum: SECOND, MINUTE, HOUR, DAY
)

tier_rate = TierRate(
    tier: str,       # Non-empty tier label (e.g., "free", "pro")
    rate: RateLimit  # Rate override for this tier
)

# Period enumeration
from domain_api_limiter import Period
period = Period.HOUR  # or SECOND, MINUTE, DAY

Policy Registry

from domain_api_limiter import PolicyRegistry

registry = PolicyRegistry()

# Retrieve the policy from a single callable
policy = registry.policy_of(my_method)  # Returns ThrottlePolicy | None

# Collect all policies from a class or module
policies = registry.collect(DocumentService)  # Returns tuple[ThrottlePolicy, ...]

Error Types

from domain_api_limiter import ThrottleError, RateLimitExceeded, ThrottleDeclarationError

# Base error for all rate-limit domain failures
# domain="api_limiter", code="throttle_error", http_status=429, retryable=True
class ThrottleError(DomainError):
    pass

# Request rejected because a rate limit was exceeded
# code="rate_limit_exceeded", http_status=429, retryable=True
class RateLimitExceeded(ThrottleError):
    pass

# Invalid throttle declaration detected at import time
# code="throttle_declaration_invalid", http_status=500, retryable=False
class ThrottleDeclarationError(ThrottleError):
    pass

All errors inherit from domain-errors DomainError and integrate with structured logging and error tracking systems.

Documentation

For detailed documentation on each feature, see:

• Throttle Policy Objects
• @throttled Decorator
• Error Types and Semantics

License

Licensed under the Apache License, Version 2.0. See LICENSE for details.

Contributing

This library is maintained by James Ekhator. Contributions welcome via pull requests. Please read CONTRIBUTING.md for guidelines and CODE_OF_CONDUCT.md for community standards.