Summary

Add asyncio.CapacityLimiter, a smarter concurrency limiter than Semaphore, adopting a proven pattern from Trio/anyio:

• Borrower tracking — tracks which tasks hold tokens; prevents same-task reacquisition (which would deadlock a semaphore)
• Dynamic total_tokens — capacity can be changed at runtime; increasing wakes waiters
• on_behalf_of API — acquire/release on behalf of arbitrary objects (enables thread integration)
• Statistics — introspection via CapacityLimiterStatistics dataclass
• WouldBlock exception — raised by nowait variants when a token isn't immediately available
• FIFO fairness — OrderedDict-based waiter queue ensures fair scheduling

Pure Python implementation, no C code needed.

Part of the proposal to adopt proven anyio/Trio patterns natively into asyncio (Tier 2, item 2.1).

API

class CapacityLimiter:
    def __init__(self, total_tokens: int | float): ...

    total_tokens: int | float          # read-write; setter wakes waiters
    borrowed_tokens: int               # read-only
    available_tokens: int | float      # read-only

    async def acquire(self) -> None
    def acquire_nowait(self) -> None
    async def acquire_on_behalf_of(self, borrower) -> None
    def acquire_on_behalf_of_nowait(self, borrower) -> None

    def release(self) -> None
    def release_on_behalf_of(self, borrower) -> None

    def statistics(self) -> CapacityLimiterStatistics

class CapacityLimiterStatistics:  # frozen dataclass
    borrowed_tokens: int
    total_tokens: int | float
    borrowers: tuple[object, ...]
    tasks_waiting: int

class WouldBlock(Exception): ...

Test plan

• Basic acquire/release cycle and context manager usage
• acquire_nowait fast path and WouldBlock when at capacity
• Same-borrower reacquisition raises RuntimeError
• Release without acquire raises RuntimeError
• FIFO fairness ordering
• Cancellation during acquire cleans up correctly
• Dynamic total_tokens increase wakes waiters
• Dynamic total_tokens decrease blocks new acquires without evicting
• Type/value validation for total_tokens
• on_behalf_of acquire/release with explicit borrowers
• statistics() reflects correct state
• repr output
• Concurrent tasks respect capacity limit
• Zero capacity and math.inf capacity edge cases
• All existing test_asyncio.test_locks tests still pass

🤖 Generated with Claude Code