"j)Z%SrSSKJr SSKJrJr SSKJr SSK J r J r SSK J r SSKJr SSKrSS KJrJr SS KJr S S KJr S rSSjrSSjr"SS\5r\ "SS55r"SS\5rSrS\S'\ S Sj5r \ S!Sj5r!S"S#Sjjr"SS.S$Sjjr#g)%zEConcurrency limiting infrastructure with OpenTelemetry observability.) annotations)ABCabstractmethod)AsyncGenerator)AbstractAsyncContextManagerasynccontextmanager) dataclass) TypeAliasN)Tracer get_tracer)Self UserError)AbstractConcurrencyLimiterConcurrencyLimiterConcurrencyLimitAnyConcurrencyLimitc.US:a[SUS35eg)Nrzmax_running must be >= 1, got z'. Use None for no concurrency limiting.r) max_runnings T/home/mande/repo/quber/.venv/lib/python3.13/site-packages/pydantic_ai/concurrency.py_validate_max_runningrs$Q8 Elmnnc6UbUS:a[SUS35egg)Nrzmax_queued must be >= 0, got z. Use None for unlimited queue.r) max_queueds r_validate_max_queuedrs,*q.7 |Cbcdd#1rc@\rSrSrSr\SSj5r\SSj5rSrg) r#aAbstract base class for concurrency limiters. Subclass this to create custom concurrency limiters (e.g., Redis-backed distributed limiters). Example: ```python from pydantic_ai.concurrency import AbstractConcurrencyLimiter class RedisConcurrencyLimiter(AbstractConcurrencyLimiter): def __init__(self, redis_client, key: str, max_running: int): self._redis = redis_client self._key = key self._max_running = max_running async def acquire(self, source: str) -> None: # Implement Redis-based distributed locking ... def release(self) -> None: # Release the Redis lock ... ``` c # g7f)znAcquire a slot, waiting if necessary. Args: source: Identifier for observability (e.g., 'model:gpt-4o'). N)selfsources racquire"AbstractConcurrencyLimiter.acquire>s  scgzRelease a slot.Nr r!s rrelease"AbstractConcurrencyLimiter.releaseGs rr Nr"strreturnNoner,r-) __name__ __module__ __qualname____firstlineno____doc__rr#r(__static_attributes__r rrrr#s/4    rrc<\rSrSr%SrS\S'SrS\S'S SjrS rg) rMaDConfiguration for concurrency limiting with optional backpressure. Args: max_running: Maximum number of concurrent operations allowed. Must be >= 1. max_queued: Maximum number of operations waiting in the queue. Must be >= 0. If None, the queue is unlimited. If exceeded, raises `ConcurrencyLimitExceeded`. intrN int | NonercX[UR5 [UR5 g)N)rrrrr's r __post_init__ConcurrencyLimit.__post_init__Zsd../T__-rr r.) r/r0r1r2r3__annotations__rr:r4r rrrrMs!J !.rrc\rSrSrSrSSSS.SSjjr\SSS.SSjj5r\SSj5r \SS j5r \SS j5r \SS j5r \SS j5r SS jrSSjrSSjrSrg)r_zA concurrency limiter that tracks waiting operations for observability. This class wraps an anyio.CapacityLimiter and tracks the number of waiting operations. When an operation has to wait to acquire a slot, a span is created for observability purposes. N)rnametracerc[U5 [U5 [R"U5UlX lX0lX@l[R"5Ul SUl g)aInitialize the ConcurrencyLimiter. Args: max_running: Maximum number of concurrent operations. Must be >= 1. max_queued: Maximum queue depth before raising ConcurrencyLimitExceeded. Must be >= 0. name: Optional name for this limiter, used for observability when sharing a limiter across multiple models or agents. tracer: OpenTelemetry tracer for span creation. Raises: UserError: If `max_running` is less than 1, or `max_queued` is less than 0. rN) rranyioCapacityLimiter_limiter _max_queued_name_tracerLock _queue_lock_waiting_count)r!rrr?r@s r__init__ConcurrencyLimiter.__init__gsL( k*Z(--k: %   ::<r)r?r@ct[U[5(aU"XUS9$U"URURUUS9$)aCCreate a ConcurrencyLimiter from a ConcurrencyLimit configuration. Args: limit: Either an int for simple limiting or a ConcurrencyLimit for full config. name: Optional name for this limiter, used for observability. tracer: OpenTelemetry tracer for span creation. Returns: A configured ConcurrencyLimiter. )rr?r@)rrr?r@) isinstancer7rr)clslimitr?r@s r from_limitConcurrencyLimiter.from_limitsC$ eS ! !5FC C!-- ++  rcUR$)z&Name of the limiter for observability.)rFr's rr?ConcurrencyLimiter.nameszzrcUR$)z9Number of operations currently waiting to acquire a slot.)rJr's r waiting_count ConcurrencyLimiter.waiting_counts"""rcJURR5R$)z'Number of operations currently running.)rD statisticsborrowed_tokensr's r running_count ConcurrencyLimiter.running_counts}}'')999rc@[URR5$)zNumber of slots available.)r7rDavailable_tokensr's ravailable_count"ConcurrencyLimiter.available_counts4==1122rc@[URR5$)z&Maximum concurrent operations allowed.)r7rD total_tokensr's rrConcurrencyLimiter.max_runnings4==--..rcJURb UR$[S5$)z9Get the tracer, falling back to global tracer if not set.z pydantic-ai)rGr r's r _get_tracerConcurrencyLimiter._get_tracers! << #<< -((rc:# SSKJn URR5 g![R a Of=fUR IShvN URbjURUR:aPUR=(d UnU"SURS-SURS3U(a SU3-5eS-5eU=RS- slSSS5IShvN O!,IShvN (df  O=fUR5nUR=(d UnUUR[URR5S .nURbURUS 'URbURUS 'S US 3nURXeS9 URR5IShvN SSS5 O!,(df  O=fU=RS-slg!U=RS-slf=f7f)zAcquire a slot, creating a span if waiting is required. Args: source: Identifier for the source of this acquisition (e.g., 'agent:my-agent' or 'model:gpt-4'). r)ConcurrencyLimitExceededNzConcurrency queue depth (z) exceeds max_queued ()z for )r"rVr limiter_namerz waiting for z concurrency) attributes) exceptionsrhrDacquire_nowaitrB WouldBlockrIrErJrFrer7rbstart_as_current_spanr#)r!r"rh display_namer@rl span_names rr#ConcurrencyLimiter.acquires 9  MM ( ( *     ###+0C0CtGWGW0W#zz3V ./0C0Ca0G/HH^_c_o_o^ppqr1=|n-GCEG   1 $ $##### %%%'F::/L !%!4!4"4==#=#=>0J zz%-1ZZ >*++/+;+; <('|nLAI--i-Omm++---POO   1 $ D  1 $ sH%H<H<HAHB C3! H,C/-H3D 9C< : D HBH-G G G H G($H+HHHc8URR5 gr&)rDr(r's rr(ConcurrencyLimiter.releases r)rDrErFrIrGrJ)rr7rr8r? str | Noner@ Tracer | None)rPzint | ConcurrencyLimitr?rvr@rwr,r )r,rv)r,r7)r,r r*r.)r/r0r1r2r3rK classmethodrQpropertyr?rVr[r_rrer#r(r4r rrrr_s"& $         <  $ %     6##::33//) 3%j rrz:int | ConcurrencyLimit | AbstractConcurrencyLimiter | Noner rc# S7v g7f)zA no-op async context manager.Nr r rr _null_contextr{s  s c# URU5IShvN S7v UR5 gN!UR5 f=f7f)zKContext manager that acquires and releases a limiter with the given source.N)r#r(limiterr"s r_limiter_contextr s= //& !!!  " s"A 3A 5A AA c2Uc [5$[X5$)a9Get an async context manager for the concurrency limiter. If limiter is None, returns a no-op context manager. Args: limiter: The AbstractConcurrencyLimiter or None. source: Identifier for the source of this acquisition (e.g., 'agent:my-agent' or 'model:gpt-4'). Returns: An async context manager. )r{rr}s rget_concurrency_contextrs G ,,rr?c^Ucg[U[5(aU$[RXS9$)aNormalize a concurrency limit configuration to an AbstractConcurrencyLimiter. Args: limit: The concurrency limit configuration. name: Optional name for the limiter if one is created. Returns: An AbstractConcurrencyLimiter if limit is not None, otherwise None. Nr)rNrrrQ)rPr?s rnormalize_to_limiterr)s3 } E5 6 6 !,,U,>>r)rr7r,r-)rr8r,r-)r,AsyncGenerator[None])r~rr"r+r,r)unnamed)r~!AbstractConcurrencyLimiter | Noner"r+r,z!AbstractAsyncContextManager[None])rPrr?rvr,r)$r3 __future__r _annotationsabcrrcollections.abcr contextlibrr dataclassesr typingr rBopentelemetry.tracer r typing_extensionsr rmr__all__rrrrrrr<r{rrrr rrrsK2#*G! 2"! o e ' ' T .. ."X 3X v"^Y]  - .- -'-.? ? ?' ?r