Tier 1

Prompt

Implement an async token bucket rate limiter in Python with the following specifications: Class AsyncTokenBucket: - __init__(self, rate: float, burst: int, initial_tokens: int | None = None) - acquire(self, tokens: int = 1) -> None (async, blocks until tokens available) - try_acquire(self, tokens: int = 1) -> bool (non-blocking) - update_rate(self, rate: float, burst: int) -> None (dynamic configuration) - get_metrics(self) -> dict (returns: tokens, rate, burst, total_acquired, total_rejected, wait_time_avg) Requirements: 1. Use asyncio.Lock for thread safety 2. Use time.monotonic() for time tracking (not time.time()) 3. Refill tokens based on elapsed time 4. Use __slots__ for memory efficiency 5. get_metrics() must return a frozen dataclass instance 6. Handle edge cases: zero rate, zero burst, negative tokens Testing requirement: Write comprehensive pytest-asyncio tests that cover: - Basic token acquisition and refill behavior - Burst limit enforcement (should not exceed burst) - Dynamic rate/burst updates taking effect immediately - Metrics accuracy (all fields correct) - Concurrent access patterns (multiple coroutines) - Edge cases (zero rate should block forever, zero burst should behave like leaky bucket, etc.) IMPORTANT: First write the test file (tests should fail initially), then implement the class to make all tests pass. Output ONLY the test file content followed by the implementation file content, separated by a clear marker like '# --- IMPLEMENTATION ---'.