RateLimiter¶

The main rate limiter classes for async and sync usage.

RateLimiter (Async)¶

RateLimiter ¶

RateLimiter(repository=None, name=None, region=None, endpoint_url=None, stack_options=None, on_unavailable=_UNSET, auto_update=_UNSET, bucket_ttl_refill_multiplier=_UNSET, speculative_writes=True)

Async rate limiter backed by DynamoDB.

Implements token bucket algorithm with support for: - Multiple limits per entity/resource - Two-level hierarchy (parent/child entities) - Cascade mode (consume from entity + parent) - Stored limit configs - Usage analytics

Example (new API - preferred): from zae_limiter import RateLimiter, Repository, StackOptions

repo = Repository(
    name="my-app",
    region="us-east-1",
    stack_options=StackOptions(),
)
limiter = RateLimiter(repository=repo)

Example (old API - deprecated): limiter = RateLimiter( name="my-app", region="us-east-1", stack_options=StackOptions(), )

Parameters:

Name	Type	Description	Default
`repository`	`RepositoryProtocol \| None`	Repository instance (new API, preferred). Pass a Repository or any RepositoryProtocol implementation.	`None`
`name`	`str \| None`	DEPRECATED. Use `Repository(name=...)` instead.	`None`
`region`	`str \| None`	DEPRECATED. Use `Repository(region=...)` instead.	`None`
`endpoint_url`	`str \| None`	DEPRECATED. Use `Repository(endpoint_url=...)` instead.	`None`
`stack_options`	`StackOptions \| None`	DEPRECATED. Use `Repository(stack_options=...)` instead.	`None`
`on_unavailable`	`OnUnavailable \| Any`	DEPRECATED. Use `set_system_defaults(on_unavailable=...)` or pass `on_unavailable=` to `acquire()` instead.	`_UNSET`
`auto_update`	`bool \| Any`	DEPRECATED. Use `Repository.builder(...).auto_update().build()` instead.	`_UNSET`
`bucket_ttl_refill_multiplier`	`int \| Any`	DEPRECATED. Use `Repository.builder(...).bucket_ttl_multiplier().build()` instead.	`_UNSET`
`speculative_writes`	`bool`	Enable speculative UpdateItem fast path. When True, acquire() tries a speculative write first, falling back to the full read-write path only when needed.	`True`

Raises:

Type	Description
`ValueError`	If both repository and name/region/endpoint_url/stack_options are provided.

name `property` ¶

name

DEPRECATED. Use repository.stack_name instead.

stack_name `property` ¶

stack_name

DEPRECATED. Use repository.stack_name instead.

table_name `property` ¶

table_name

DEPRECATED. Use repository.stack_name instead.

list_deployed `async` `classmethod` ¶

list_deployed(region=None, endpoint_url=None)

List all deployed rate limiter instances in a region.

This is a class method that discovers existing deployments without requiring an initialized RateLimiter instance. It queries CloudFormation for stacks tagged with ManagedBy=zae-limiter.

Parameters:

Name	Type	Description	Default
`region`	`str \| None`	AWS region (default: use boto3 defaults)	`None`
`endpoint_url`	`str \| None`	CloudFormation endpoint (for LocalStack)	`None`

Returns:

Type	Description
`list[LimiterInfo]`	List of LimiterInfo objects describing deployed instances.
`list[LimiterInfo]`	Sorted by user-friendly name. Excludes deleted stacks.

Example

Discover all limiters in us-east-1¶

limiters = await RateLimiter.list_deployed(region="us-east-1") for limiter in limiters: if limiter.is_healthy: print(f"✓ {limiter.user_name}: {limiter.version}") elif limiter.is_failed: print(f"✗ {limiter.user_name}: {limiter.stack_status}")

Raises:

Type	Description
`ClientError`	If CloudFormation API call fails

close `async` ¶

close()

Close the underlying connections.

is_available `async` ¶

is_available(timeout=1.0)

Check if the rate limiter backend (DynamoDB) is reachable.

Performs a lightweight health check without requiring initialization. This method never raises exceptions - it returns False on any error.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Maximum time in seconds to wait for response (default: 1.0)	`1.0`

Returns:

Type	Description
`bool`	True if DynamoDB table is reachable, False otherwise.

Example

limiter = RateLimiter(name="my-app", region="us-east-1") if await limiter.is_available(): async with limiter.acquire(...) as lease: ... else: # Handle degraded mode pass

create_entity `async` ¶

create_entity(entity_id, name=None, parent_id=None, cascade=False, metadata=None, principal=None)

Create a new entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Unique identifier for the entity	required
`name`	`str \| None`	Human-readable name (defaults to entity_id)	`None`
`parent_id`	`str \| None`	Parent entity ID (None for root/project entities)	`None`
`cascade`	`bool`	If True, acquire() will also consume from parent entity	`False`
`metadata`	`dict[str, str] \| None`	Additional metadata to store	`None`
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

Returns:

Type	Description
`Entity`	The created Entity

Raises:

Type	Description
`EntityExistsError`	If entity already exists

get_entity `async` ¶

get_entity(entity_id)

Get an entity by ID.

delete_entity `async` ¶

delete_entity(entity_id, principal=None)

Delete an entity and all its related data.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	ID of the entity to delete	required
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

Raises:

Type	Description
`RateLimiterUnavailable`	DynamoDB kept some deletes unprocessed after the retries. What was deleted is audited, and calling `delete_entity` again finishes the rest.

get_children `async` ¶

get_children(parent_id)

Get all children of a parent entity.

set_parent `async` ¶

set_parent(entity_id, parent_id, *, principal=None)

Move an entity to a new parent, or to no parent (ADR-150).

Takes effect on the next acquire in every process: the entity's bucket items are restamped, and a process whose cache still names the old parent learns the new one from the item on its next call, refunding the old parent. Consumption already charged to the old parent, including by leases open across the move, stays there. Whether the entity debits the new parent follows its cascade policy (ADR-146) or its own cascade flag, which a move does not change.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to move.	required
`parent_id`	`str \| None`	The new parent, which must exist, or None for no parent.	required
`principal`	`str \| None`	Caller identity for audit logging (optional).	`None`

Returns:

Type	Description
`int`	Number of bucket items restamped.

Raises:

Type	Description
`ValidationError`	The move would make the entity its own ancestor.
`EntityNotFoundError`	The entity or the new parent does not exist.
`VersionMismatchError`	The stack's Lambdas predate parent moves.
`FanoutIncomplete`	The move landed but restamping stopped part-way; repeating the call finishes it.

get_audit_events `async` ¶

get_audit_events(entity_id, limit=100, start_event_id=None)

Get audit events for an entity.

Retrieves security audit events logged for administrative operations on the specified entity, ordered by most recent first.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	ID of the entity to query	required
`limit`	`int`	Maximum number of events to return (default: 100)	`100`
`start_event_id`	`str \| None`	Event ID to start after (for pagination)	`None`

Returns:

Type	Description
`list[AuditEvent]`	List of AuditEvent objects, ordered by most recent first

Example

events = await limiter.get_audit_events("proj-1") for event in events: print(f"{event.timestamp}: {event.action} by {event.principal}")

get_usage_snapshots `async` ¶

get_usage_snapshots(entity_id=None, resource=None, window_type=None, start_time=None, end_time=None, limit=100, next_key=None)

Query usage snapshots for historical consumption data.

Usage snapshots are created by the aggregator Lambda from DynamoDB stream events. They track token consumption per entity/resource within time windows (hourly, daily).

Supports two query modes: 1. Entity-scoped: Provide entity_id (optionally with resource filter) 2. Resource-scoped: Provide resource to query across all entities

Parameters:

Name	Type	Description	Default
`entity_id`	`str \| None`	Entity to query (uses primary key)	`None`
`resource`	`str \| None`	Resource name filter (required if entity_id is None)	`None`
`window_type`	`str \| None`	Filter by window type ("hourly", "daily")	`None`
`start_time`	`datetime \| None`	Filter snapshots >= this timestamp	`None`
`end_time`	`datetime \| None`	Filter snapshots <= this timestamp	`None`
`limit`	`int`	Maximum items to fetch from DynamoDB per page (default: 100)	`100`
`next_key`	`dict[str, Any] \| None`	Pagination cursor from previous call	`None`

Returns:

Type	Description
`tuple[list[UsageSnapshot], dict[str, Any] \| None]`	Tuple of (snapshots, next_key). next_key is None if no more results.

Raises:

Type	Description
`ValueError`	If neither entity_id nor resource is provided

Note

The limit parameter controls the DynamoDB query batch size. Client-side filters (window_type, start_time, end_time) are applied after fetching, so the returned count may be less than limit. Use next_key to paginate through all matching results.

Example

Get hourly snapshots for an entity¶

snapshots, cursor = await limiter.get_usage_snapshots( entity_id="user-123", resource="gpt-4", window_type="hourly", start_time=datetime(2024, 1, 1), end_time=datetime(2024, 1, 31), ) for snap in snapshots: print(f"{snap.window_start}: {snap.counters}")

Paginate through results¶

while cursor: more, cursor = await limiter.get_usage_snapshots( entity_id="user-123", next_key=cursor, )

get_usage_summary `async` ¶

get_usage_summary(entity_id=None, resource=None, window_type=None, start_time=None, end_time=None)

Get aggregated usage summary across multiple snapshots.

Fetches all matching snapshots and computes total and average consumption statistics. Useful for billing, reporting, and capacity planning.

Parameters:

Name	Type	Description	Default
`entity_id`	`str \| None`	Entity to query	`None`
`resource`	`str \| None`	Resource name filter (required if entity_id is None)	`None`
`window_type`	`str \| None`	Filter by window type ("hourly", "daily")	`None`
`start_time`	`datetime \| None`	Filter snapshots >= this timestamp	`None`
`end_time`	`datetime \| None`	Filter snapshots <= this timestamp	`None`

Returns:

Type	Description
`UsageSummary`	UsageSummary with total and average consumption per limit type

Raises:

Type	Description
`ValueError`	If neither entity_id nor resource is provided

Example

summary = await limiter.get_usage_summary( entity_id="user-123", resource="gpt-4", window_type="hourly", start_time=datetime(2024, 1, 1), end_time=datetime(2024, 1, 31), ) print(f"Total tokens: {summary.total.get('tpm', 0)}") print(f"Average per hour: {summary.average.get('tpm', 0.0):.1f}") print(f"Snapshots: {summary.snapshot_count}")

acquire `async` ¶

acquire(entity_id, resource, consume, limits=None, use_stored_limits=False, on_unavailable=None, *, also=None)

Acquire rate limit capacity.

Limits are resolved automatically from stored config using four-tier hierarchy: Entity > Entity Default > Resource > System. Pass limits to override.

Cascade behavior is controlled by the entity's cascade flag, set at entity creation time via create_entity(cascade=True). When enabled, acquire() automatically consumes from both the entity and its parent.

Several resources can be debited in one acquire, all or none (ADR-148): name the extra ones in also. Each resolves its own limits, disabled flag, cascade policy, shards and quota grants exactly as a single-resource acquire does. When any resource is rejected, disabled or unavailable, every debit the acquire wrote is refunded before the exception reaches the caller. The yielded lease covers every resource: its own adjust(), consume(), release() and consumed act on resource, and lease.resource(name) reaches each of the others.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to acquire capacity for	required
`resource`	`str`	Resource being accessed (e.g., "gpt-4")	required
`consume`	`dict[str, int]`	Amounts to consume by limit name, each >= 0 (return capacity with `lease.release()`)	required
`limits`	`list[Limit] \| None`	Override stored config with explicit limits (optional). Cannot be combined with `also`.	`None`
`use_stored_limits`	`bool`	DEPRECATED - limits are now always resolved from stored config. This parameter will be removed in v1.0.	`False`
`on_unavailable`	`OnUnavailable \| None`	Override default on_unavailable behavior	`None`
`also`	`dict[str, dict[str, int]] \| None`	Further resources to debit in the same acquire, mapping each resource name to its amounts by limit name (optional). At most `MAX_ACQUIRE_RESOURCES - 1` entries, none naming `resource`.	`None`

Yields:

Type	Description
`AsyncIterator[Lease]`	Lease for managing additional consumption

Raises:

Type	Description
`RateLimitExceeded`	If any limit would be exceeded. With `also`, carries a status for every declared limit of every resource the decision evaluated, each tagged with its resource.
`ResourceDisabled`	If the resource — or, with `also`, any of the resources — is disabled. Outranks `RateLimitExceeded`.
`RateLimiterUnavailable`	If DynamoDB unavailable and BLOCK
`ValidationError`	If no limits configured at any level, an amount in `consume` or `also` is negative, or `also` is invalid (a bad or repeated resource name, too many resources, or combined with `limits`)
`VersionMismatchError`	If `limits` carries a `reset_after` limit and the stack's Lambdas predate it (#638). Never subject to `on_unavailable`.

check_availability `async` ¶

check_availability(entity_id, resource, needed=None, limits=None)

Check available capacity and wait time in a single read (issue #472).

Answers both "how much is left?" and "how long until I can proceed?" for every resolved limit, from one config resolution and one bucket read taken at one instant.

Why one call rather than two. Calling :meth:available and :meth:time_until_available back to back is two reads at two instants, and they can disagree: tokens refill in between, and each discovers the entity's shards separately. A display built from the pair can render "0 remaining, available now" or "47 remaining, wait 0s". Both are lies a user acts on. Everything on the returned :class:Availability is derived from a single snapshot stamped with checked_at_ms, so the numbers cannot contradict each other.

Limits are resolved using four-tier hierarchy: Entity > Entity Default > Resource > System. If no stored limits found, falls back to the limits parameter.

Consumes nothing and writes nothing. This is a read; it is not a pre-flight check for :meth:acquire. Deciding with it and then calling acquire() is TOCTOU and costs an extra read — acquire() already answers "may I proceed, and if not when" in 1 WCU, with no read even on a fast rejection, via RateLimitExceeded.retry_after_seconds. Use this when the answer is displayed rather than acted on.

Sharding. A sharded entity's balance is spread across its shards (GHSA-76rv), so this sums every shard, exactly as :meth:available and get_resource_capacity() do, and computes the wait against the summed refill rate. The reported limit is therefore the undivided configured limit, not one shard's share — unlike the per-shard statuses in RateLimitExceeded (#475), which describe the one shard a rejection happened on. Known limitation, inherited from #475: a single request larger than capacity // shard_count is unadmittable on every shard even while the entity is under its configured limit, so acquire() can reject an amount this call reports as available.

Schedules. Everything reported is the value in force at checked_at_ms, not the stored base (#222 §7): the ceiling comes from effective_params, so a scale: 0.5 window reports 500 and not 1000, and the wait walks forward across boundaries instead of dividing by the rate that happens to apply right now. A limit with a reset_schedule reports the wait to its next edge — for a daily quota, whose refill_amount is 0 by ADR-137, that is the only finite answer there is. A bucket that crossed a reset edge and has not been written to since reports the balance the next acquire() will restore, per shard, rather than the burnt one still on disk.

Cost: 1 GSI3 query (KEYS_ONLY) + 1 BatchGetItem, plus config resolution (free on a cache hit), regardless of limit count or shard count. A limit carrying a reset_schedule adds one backwards cron scan per shard and the forward walk adds up to eight per exceeded limit; an unscheduled limit adds neither.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to check	required
`resource`	`str`	Resource to check	required
`needed`	`dict[str, int] \| None`	Required amounts by limit name. Omit to ask only about current availability (`retry_after_seconds` is then 0.0); pass e.g. `{"rpm": 1}` for "when may I make one more request". Keys naming no resolved limit are ignored.	`None`
`limits`	`list[Limit] \| None`	Override limits (optional, falls back to stored config)	`None`

Returns:

Type	Description
`Availability`	Availability carrying one :class:`LimitStatus` per resolved limit,
`Availability`	plus the derived aggregate verdict and countdown.

Raises:

Type	Description
`ValidationError`	If no limits found at any level and no override provided

Example

check = await limiter.check_availability(
    entity_id="key-abc",
    resource="gpt-4",
    needed={"rpm": 1, "tpm": 500},
)
for status in check.statuses:
    render(status.limit_name, status.available, status.retry_after_seconds)

available `async` ¶

available(entity_id, resource, limits=None, use_stored_limits=False)

Check available capacity without consuming.

Limits are resolved using four-tier hierarchy: Entity > Entity Default > Resource > System. If no stored limits found, falls back to the limits parameter.

Sums every shard of the entity (GHSA-76rv). Can return negative values if the bucket is in debt.

Thin wrapper over :meth:check_availability, which returns this and the wait time together from one read at one instant. Prefer it when you want both — asking here and there is two snapshots that can disagree.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to check	required
`resource`	`str`	Resource to check	required
`limits`	`list[Limit] \| None`	Override limits (optional, falls back to stored config)	`None`
`use_stored_limits`	`bool`	DEPRECATED - limits are now always resolved from stored config. This parameter will be removed in v1.0.	`False`

Returns:

Type	Description
`dict[str, int]`	Dict mapping limit_name -> available tokens

Raises:

Type	Description
`ValidationError`	If no limits found at any level and no override provided

time_until_available `async` ¶

time_until_available(entity_id, resource, needed, limits=None, use_stored_limits=False)

Calculate seconds until requested capacity is available.

Limits are resolved using four-tier hierarchy: Entity > Entity Default > Resource > System. If no stored limits found, falls back to the limits parameter.

The estimate is computed against the entity's total across every shard and the summed refill rate (GHSA-76rv), matching :meth:available.

Thin wrapper over :meth:check_availability, which returns this and the available capacity together from one read at one instant. Prefer it when you want both, or when you want the wait per limit rather than the slowest one.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to check	required
`resource`	`str`	Resource to check	required
`needed`	`dict[str, int]`	Required amounts by limit name	required
`limits`	`list[Limit] \| None`	Override limits (optional, falls back to stored config)	`None`
`use_stored_limits`	`bool`	DEPRECATED - limits are now always resolved from stored config. This parameter will be removed in v1.0.	`False`

Returns:

Type	Description
`float`	Seconds until available (0.0 if already available)

Raises:

Type	Description
`ValidationError`	If no limits found at any level and no override provided

set_limits `async` ¶

set_limits(entity_id, limits, resource=DEFAULT_RESOURCE, principal=None)

Store limit configs for an entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to set limits for	required
`limits`	`list[Limit]`	Limits to store	required
`resource`	`str`	Resource these limits apply to (or default)	`DEFAULT_RESOURCE`
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

get_limits `async` ¶

get_limits(entity_id, resource=DEFAULT_RESOURCE)

Get stored limit configs for an entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to get limits for	required
`resource`	`str`	Resource to get limits for	`DEFAULT_RESOURCE`

Returns:

Type	Description
`list[Limit]`	List of stored limits (empty if none)

delete_limits `async` ¶

delete_limits(entity_id, resource=DEFAULT_RESOURCE, principal=None)

Delete stored limit configs for an entity.

Reconciles existing buckets to fallback config (resource/system defaults) by syncing limit fields, setting TTL, and removing stale limit attributes (issue #327).

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to delete limits for	required
`resource`	`str`	Resource to delete limits for	`DEFAULT_RESOURCE`
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

list_entities_with_custom_limits `async` ¶

list_entities_with_custom_limits(resource, limit=None, cursor=None)

List all entities that have custom limit configurations.

Uses GSI3 sparse index for efficient queries. Only entities with custom limits for the specified resource are returned.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Resource to filter by.	required
`limit`	`int \| None`	Maximum number of entities to return. None for all.	`None`
`cursor`	`str \| None`	Pagination cursor from previous call.	`None`

Returns:

Type	Description
`tuple[list[str], str \| None]`	Tuple of (entity_ids, next_cursor). next_cursor is None if no more results.

Example

Get all entities with custom limits for gpt-4¶

entities, cursor = await limiter.list_entities_with_custom_limits("gpt-4") for entity_id in entities: print(entity_id)

Paginate through results¶

while cursor: more, cursor = await limiter.list_entities_with_custom_limits( "gpt-4", cursor=cursor ) entities.extend(more)

list_resources_with_entity_configs `async` ¶

list_resources_with_entity_configs()

List all resources that have entity-level custom limit configurations.

Uses the entity config resources registry for efficient O(1) lookup.

Returns:

Type	Description
`list[str]`	Sorted list of resource names with at least one entity having custom limits

Example

resources = await limiter.list_resources_with_entity_configs() for resource in resources: entities, _ = await limiter.list_entities_with_custom_limits(resource) print(f"{resource}: {len(entities)} entities with custom limits")

set_resource_defaults `async` ¶

set_resource_defaults(resource, limits, principal=None)

Store default limit configs for a resource.

Resource defaults override system defaults for the specified resource.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Resource name	required
`limits`	`list[Limit]`	Limits to store	required
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

get_resource_defaults `async` ¶

get_resource_defaults(resource)

Get stored default limit configs for a resource.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Resource name	required

Returns:

Type	Description
`list[Limit]`	List of stored limits (empty if none)

delete_resource_defaults `async` ¶

delete_resource_defaults(resource, principal=None)

Delete stored default limit configs for a resource.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Resource name	required
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

list_resources_with_defaults `async` ¶

list_resources_with_defaults()

List all resources that have default limit configs.

set_system_defaults `async` ¶

set_system_defaults(limits, on_unavailable=None, principal=None)

Store system-wide default limits and config.

System defaults apply to ALL resources unless overridden at resource or entity level.

Parameters:

Name	Type	Description	Default
`limits`	`list[Limit]`	Limits to store (apply globally to all resources)	required
`on_unavailable`	`OnUnavailable \| None`	Behavior when DynamoDB unavailable (optional)	`None`
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

get_system_defaults `async` ¶

get_system_defaults()

Get system-wide default limits and config.

Returns:

Type	Description
`tuple[list[Limit], OnUnavailable \| None]`	Tuple of (limits, on_unavailable). on_unavailable may be None if not set.

delete_system_defaults `async` ¶

delete_system_defaults(principal=None)

Delete all system-wide default limits and config.

Parameters:

Name	Type	Description	Default
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

get_resource_capacity `async` ¶

get_resource_capacity(resource, limit_name, parents_only=False)

Get aggregated capacity for a resource across all entities.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Resource to query	required
`limit_name`	`str`	Limit name to query	required
`parents_only`	`bool`	If True, only include parent entities	`False`

Returns:

Type	Description
`ResourceCapacity`	ResourceCapacity with aggregated data

SyncRateLimiter¶

SyncRateLimiter ¶

SyncRateLimiter(repository=None, name=None, region=None, endpoint_url=None, stack_options=None, on_unavailable=_UNSET, auto_update=_UNSET, bucket_ttl_refill_multiplier=_UNSET, speculative_writes=True)

Async rate limiter backed by DynamoDB.

Implements token bucket algorithm with support for: - Multiple limits per entity/resource - Two-level hierarchy (parent/child entities) - Cascade mode (consume from entity + parent) - Stored limit configs - Usage analytics

Example (new API - preferred): from zae_limiter import SyncRateLimiter, SyncRepository, StackOptions

repo = SyncRepository(
    name="my-app",
    region="us-east-1",
    stack_options=StackOptions(),
)
limiter = SyncRateLimiter(repository=repo)

Example (old API - deprecated): limiter = SyncRateLimiter( name="my-app", region="us-east-1", stack_options=StackOptions(), )

Parameters:

Name	Type	Description	Default
`repository`	`SyncRepositoryProtocol \| None`	SyncRepository instance (new API, preferred). Pass a SyncRepository or any SyncRepositoryProtocol implementation.	`None`
`name`	`str \| None`	DEPRECATED. Use `SyncRepository(name=...)` instead.	`None`
`region`	`str \| None`	DEPRECATED. Use `SyncRepository(region=...)` instead.	`None`
`endpoint_url`	`str \| None`	DEPRECATED. Use `SyncRepository(endpoint_url=...)` instead.	`None`
`stack_options`	`StackOptions \| None`	DEPRECATED. Use `SyncRepository(stack_options=...)` instead.	`None`
`on_unavailable`	`OnUnavailable \| Any`	DEPRECATED. Use `set_system_defaults(on_unavailable=...)` or pass `on_unavailable=` to `acquire()` instead.	`_UNSET`
`auto_update`	`bool \| Any`	DEPRECATED. Use `SyncRepository.builder(...).auto_update().build()` instead.	`_UNSET`
`bucket_ttl_refill_multiplier`	`int \| Any`	DEPRECATED. Use `SyncRepository.builder(...).bucket_ttl_multiplier().build()` instead.	`_UNSET`
`speculative_writes`	`bool`	Enable speculative UpdateItem fast path. When True, acquire() tries a speculative write first, falling back to the full read-write path only when needed.	`True`

Raises:

Type	Description
`ValueError`	If both repository and name/region/endpoint_url/stack_options are provided.

name `property` ¶

name

DEPRECATED. Use repository.stack_name instead.

stack_name `property` ¶

stack_name

DEPRECATED. Use repository.stack_name instead.

table_name `property` ¶

table_name

DEPRECATED. Use repository.stack_name instead.

list_deployed `classmethod` ¶

list_deployed(region=None, endpoint_url=None)

List all deployed rate limiter instances in a region.

This is a class method that discovers existing deployments without requiring an initialized SyncRateLimiter instance. It queries CloudFormation for stacks tagged with ManagedBy=zae-limiter.

Parameters:

Name	Type	Description	Default
`region`	`str \| None`	AWS region (default: use boto3 defaults)	`None`
`endpoint_url`	`str \| None`	CloudFormation endpoint (for LocalStack)	`None`

Returns:

Type	Description
`list[LimiterInfo]`	List of LimiterInfo objects describing deployed instances.
`list[LimiterInfo]`	Sorted by user-friendly name. Excludes deleted stacks.

Example

Discover all limiters in us-east-1¶

limiters = SyncRateLimiter.list_deployed(region="us-east-1") for limiter in limiters: if limiter.is_healthy: print(f"✓ {limiter.user_name}: {limiter.version}") elif limiter.is_failed: print(f"✗ {limiter.user_name}: {limiter.stack_status}")

Raises:

Type	Description
`ClientError`	If CloudFormation API call fails

close ¶

close()

Close the underlying connections.

is_available ¶

is_available(timeout=1.0)

Check if the rate limiter backend (DynamoDB) is reachable.

Performs a lightweight health check without requiring initialization. This method never raises exceptions - it returns False on any error.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Maximum time in seconds to wait for response (default: 1.0)	`1.0`

Returns:

Type	Description
`bool`	True if DynamoDB table is reachable, False otherwise.

Example

limiter = SyncRateLimiter(name="my-app", region="us-east-1") if limiter.is_available(): async with limiter.acquire(...) as lease: ... else: # Handle degraded mode pass

create_entity ¶

create_entity(entity_id, name=None, parent_id=None, cascade=False, metadata=None, principal=None)

Create a new entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Unique identifier for the entity	required
`name`	`str \| None`	Human-readable name (defaults to entity_id)	`None`
`parent_id`	`str \| None`	Parent entity ID (None for root/project entities)	`None`
`cascade`	`bool`	If True, acquire() will also consume from parent entity	`False`
`metadata`	`dict[str, str] \| None`	Additional metadata to store	`None`
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

Returns:

Type	Description
`Entity`	The created Entity

Raises:

Type	Description
`EntityExistsError`	If entity already exists

get_entity ¶

get_entity(entity_id)

Get an entity by ID.

delete_entity ¶

delete_entity(entity_id, principal=None)

Delete an entity and all its related data.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	ID of the entity to delete	required
`principal`	`str \| None`	Caller identity for audit logging (optional)	`None`

Raises:

Type	Description
`RateLimiterUnavailable`	DynamoDB kept some deletes unprocessed after the retries. What was deleted is audited, and calling `delete_entity` again finishes the rest.

get_children ¶

get_children(parent_id)

Get all children of a parent entity.

set_parent ¶

set_parent(entity_id, parent_id, *, principal=None)

Move an entity to a new parent, or to no parent (ADR-150).

Takes effect on the next acquire in every process: the entity's bucket items are restamped, and a process whose cache still names the old parent learns the new one from the item on its next call, refunding the old parent. Consumption already charged to the old parent, including by leases open across the move, stays there. Whether the entity debits the new parent follows its cascade policy (ADR-146) or its own cascade flag, which a move does not change.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to move.	required
`parent_id`	`str \| None`	The new parent, which must exist, or None for no parent.	required
`principal`	`str \| None`	Caller identity for audit logging (optional).	`None`

Returns:

Type	Description
`int`	Number of bucket items restamped.

Raises:

Type	Description
`ValidationError`	The move would make the entity its own ancestor.
`EntityNotFoundError`	The entity or the new parent does not exist.
`VersionMismatchError`	The stack's Lambdas predate parent moves.
`FanoutIncomplete`	The move landed but restamping stopped part-way; repeating the call finishes it.

get_audit_events ¶

get_audit_events(entity_id, limit=100, start_event_id=None)

Get audit events for an entity.

Retrieves security audit events logged for administrative operations on the specified entity, ordered by most recent first.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	ID of the entity to query	required
`limit`	`int`	Maximum number of events to return (default: 100)	`100`
`start_event_id`	`str \| None`	Event ID to start after (for pagination)	`None`

Returns:

Type	Description
`list[AuditEvent]`	List of AuditEvent objects, ordered by most recent first

Example

events = limiter.get_audit_events("proj-1") for event in events: print(f"{event.timestamp}: {event.action} by {event.principal}")

get_usage_snapshots ¶

get_usage_snapshots(entity_id=None, resource=None, window_type=None, start_time=None, end_time=None, limit=100, next_key=None)

Query usage snapshots for historical consumption data.

Usage snapshots are created by the aggregator Lambda from DynamoDB stream events. They track token consumption per entity/resource within time windows (hourly, daily).

Supports two query modes: 1. Entity-scoped: Provide entity_id (optionally with resource filter) 2. Resource-scoped: Provide resource to query across all entities

Parameters:

Name	Type	Description	Default
`entity_id`	`str \| None`	Entity to query (uses primary key)	`None`
`resource`	`str \| None`	Resource name filter (required if entity_id is None)	`None`
`window_type`	`str \| None`	Filter by window type ("hourly", "daily")	`None`
`start_time`	`datetime \| None`	Filter snapshots >= this timestamp	`None`
`end_time`	`datetime \| None`	Filter snapshots <= this timestamp	`None`
`limit`	`int`	Maximum items to fetch from DynamoDB per page (default: 100)	`100`
`next_key`	`dict[str, Any] \| None`	Pagination cursor from previous call	`None`

Returns:

Type	Description
`tuple[list[UsageSnapshot], dict[str, Any] \| None]`	Tuple of (snapshots, next_key). next_key is None if no more results.

Raises:

Type	Description
`ValueError`	If neither entity_id nor resource is provided

Note

The limit parameter controls the DynamoDB query batch size. Client-side filters (window_type, start_time, end_time) are applied after fetching, so the returned count may be less than limit. Use next_key to paginate through all matching results.

Example

Get hourly snapshots for an entity¶

snapshots, cursor = limiter.get_usage_snapshots( entity_id="user-123", resource="gpt-4", window_type="hourly", start_time=datetime(2024, 1, 1), end_time=datetime(2024, 1, 31), ) for snap in snapshots: print(f"{snap.window_start}: {snap.counters}")

Paginate through results¶

while cursor: more, cursor = limiter.get_usage_snapshots( entity_id="user-123", next_key=cursor, )

get_usage_summary ¶

get_usage_summary(entity_id=None, resource=None, window_type=None, start_time=None, end_time=None)

Get aggregated usage summary across multiple snapshots.

Fetches all matching snapshots and computes total and average consumption statistics. Useful for billing, reporting, and capacity planning.

Parameters:

Name	Type	Description	Default
`entity_id`	`str \| None`	Entity to query	`None`
`resource`	`str \| None`	Resource name filter (required if entity_id is None)	`None`
`window_type`	`str \| None`	Filter by window type ("hourly", "daily")	`None`
`start_time`	`datetime \| None`	Filter snapshots >= this timestamp	`None`
`end_time`	`datetime \| None`	Filter snapshots <= this timestamp	`None`

Returns:

Type	Description
`UsageSummary`	UsageSummary with total and average consumption per limit type

Raises:

Type	Description
`ValueError`	If neither entity_id nor resource is provided

Example

summary = limiter.get_usage_summary( entity_id="user-123", resource="gpt-4", window_type="hourly", start_time=datetime(2024, 1, 1), end_time=datetime(2024, 1, 31), ) print(f"Total tokens: {summary.total.get('tpm', 0)}") print(f"Average per hour: {summary.average.get('tpm', 0.0):.1f}") print(f"Snapshots: {summary.snapshot_count}")

acquire ¶

acquire(entity_id, resource, consume, limits=None, use_stored_limits=False, on_unavailable=None, *, also=None)

Acquire rate limit capacity.

Limits are resolved automatically from stored config using four-tier hierarchy: Entity > Entity Default > Resource > System. Pass limits to override.

Cascade behavior is controlled by the entity's cascade flag, set at entity creation time via create_entity(cascade=True). When enabled, acquire() automatically consumes from both the entity and its parent.

Several resources can be debited in one acquire, all or none (ADR-148): name the extra ones in also. Each resolves its own limits, disabled flag, cascade policy, shards and quota grants exactly as a single-resource acquire does. When any resource is rejected, disabled or unavailable, every debit the acquire wrote is refunded before the exception reaches the caller. The yielded lease covers every resource: its own adjust(), consume(), release() and consumed act on resource, and lease.resource(name) reaches each of the others.

Parameters:

Name	Type	Description	Default
`entity_id`	`str`	Entity to acquire capacity for	required
`resource`	`str`	Resource being accessed (e.g., "gpt-4")	required
`consume`	`dict[str, int]`	Amounts to consume by limit name, each >= 0 (return capacity with `lease.release()`)	required
`limits`	`list[Limit] \| None`	Override stored config with explicit limits (optional). Cannot be combined with `also`.	`None`
`use_stored_limits`	`bool`	DEPRECATED - limits are now always resolved from stored config. This parameter will be removed in v1.0.	`False`
`on_unavailable`	`OnUnavailable \| None`	Override default on_unavailable behavior	`None`
`also`	`dict[str, dict[str, int]] \| None`	Further resources to debit in the same acquire, mapping each resource name to its amounts by limit name (optional). At most `MAX_ACQUIRE_RESOURCES - 1` entries, none naming `resource`.	`None`

Yields:

Type	Description
`SyncLease`	SyncLease for managing additional consumption

Raises:

Type	Description
`RateLimitExceeded`	If any limit would be exceeded. With `also`, carries a status for every declared limit of every resource the decision evaluated, each tagged with its resource.
`ResourceDisabled`	If the resource — or, with `also`, any of the resources — is disabled. Outranks `RateLimitExceeded`.
`RateLimiterUnavailable`	If DynamoDB unavailable and BLOCK
`ValidationError`	If no limits configured at any level, an amount in `consume` or `also` is negative, or `also` is invalid (a bad or repeated resource name, too many resources, or combined with `limits`)
`VersionMismatchError`	If `limits` carries a `reset_after` limit and the stack's Lambdas predate it (#638). Never subject to `on_unavailable`.

check_availability ¶

check_availability(entity_id, resource, needed=None, limits=None)