rPj)K DUdZddlZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z m Z mZddlZddlmZdd lmZdd lmZmZmZmZej2eZd Z d Zd Zd ZdZ dZ!dZ"iZ#e$e%e&e'e&zdze&fe%dejPffe)d<ejTZ+Gdde,Z-eGddZ.eGddZ/eGddZ0eGddZ1eGddZ2ee/e0e1e2fZ3de d e dzfd!Z4d"e&d#e&d e3dzfd$Z5e%e d%e&fZ6d&e6dzd e$e&e&ffd'Z7d(ee8d)ege'fd ee8fd*Z9d+e&d,e&d-e'e&zdzd&e6dzd)ege'fd ee3f d.Z:Gd/d0Z;d1e,d edzfd3Z?Gd4dejZAy)5a*Live-following of file changes in a bucket. The Hub exposes a server-sent events feed of per-file changes for buckets: ``` GET /api/buckets/{bucket_id}/events?cursor= | ?since= Accept: text/event-stream ``` A consumer can keep its view of a bucket fresh by applying the batches as they happen instead of re-listing directories. This module implements the client side of that feed: - [`parse_follow_event`] turns a raw SSE event into a typed event. Unknown or malformed events are dropped so the Hub can add event types without breaking older clients. - [`BucketFollower`] is a daemon thread owning one (re)connecting stream for one bucket. It maintains the resume point, reconnects with exponential backoff, and hands the batches over to callbacks provided by the caller (see [`HfFileSystem`]). The feed is best-effort by design: when it cannot be followed (deployment where live-follow is disabled, token without access to the bucket), the follower stops and the consumer is left with its default behavior. Event vocabulary: | event | payload | meaning | | ----------- | -------------------------------------- | --------------------------------------------------------------------- | | `ready` | `{cursor?}` | replay is done, the stream now follows live changes | | `changes` | `{cursor, changes: [{path, op, ...}]}` | batch of per-file changes; `cursor` resumes strictly after the batch | | `reset` | `{reason}` | resume point older than the replay buffer: reconcile and re-subscribe | | `reconnect` | `{cursor?}` | server-directed end of stream (pod rotation, shutdown,...) | N)CallableIterator) dataclassfield)datetime)AnyLiteralUnion)logging) SSEClient)build_hf_headers get_sessionhf_raise_for_statusparse_datetimegV@$@gT@g?)iiiiBucketFollowerBUCKET_FOLLOWERSceZdZdZy)BucketEventsUnavailablezFThe live-follow feed cannot be followed for this bucket: do not retry.N)__name__ __module__ __qualname____doc__h/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/huggingface_hub/_bucket_live_follow.pyrrRsPrrc|eZdZUdZeed<eded<dZedzed<dZ edzed<dZ e dzed<dZ e dzed <y) BucketFileChangeaQ A single file change reported by the live-follow feed of a bucket. Only `path` and `op` are always set: an `update` carries only the fields that changed, so a field left to `None` means "unchanged" (or, for `xet_hash`, possibly "not readable with this token") and never "cleared". Args: path (`str`): Path of the file in the bucket. op (`Literal["add", "update", "delete"]`): What happened to the file. size (`int`, *optional*): Size of the file in bytes, when reported. xet_hash (`str`, *optional*): Xet hash of the file, when reported. uploaded_at (`datetime`, *optional*): Upload instant, set on every add and re-upload. mtime (`datetime`, *optional*): Client-provided modification time, when reported. pathaddupdatedeleteopNsizexet_hash uploaded_atmtime) rrrrstr__annotations__r r'intr(r)rr*rrrr r VsR* I)**D#*HcDj#'KD'!E8d?!rr ceZdZUedzed<y)_ReadyNcursorrrrr+r,rrrr/r/u $Jrr/c:eZdZUeed<eeZeeed<y)_Changesr0)default_factorychangesN) rrrr+r,rlistr6r rrrr4r4zs K&+D&AGT" #Arr4c eZdZy)_ResetN)rrrrrrr9r9srr9ceZdZUedzed<y) _ReconnectNr0r1rrrr;r;r2rr;valuereturnc@t|tr |r t|SdSN) isinstancer+r)r<s r_parse_datetimerAs$.uc$:u>% N$Nrnamedatac |jrtj|ni}t |tstj d||ddy|xdk(r5|jd}tt |tr|SdSxdk(r|jd|jd}}t |trt |tstj d |ddyg}|D]}t |tr|jd d vr'|jt|jd d |d |jd|jdt|jdt|jdt||Sxdk(r t!Sdk(r4|jd}t#t |tr|SdS tj d|y#tj$rtj d||ddYywxYw)a Parse a raw live-follow event into a typed event. Args: name (`str`): Value of the `event:` field: `ready`, `changes`, `reset` or `reconnect`. data (`str`): JSON payload of the `data:` field (may be empty). Returns: `Any` | `None`: The parsed event, or `None` when it must be ignored (unknown event name or malformed payload). z.live-follow: ignoring malformed %r payload: %sNz:live-follow: ignoring %r payload that is not an object: %sreadyr0r0r6z@live-follow: ignoring 'changes' event without cursor/changes: %sr&r"r!r'xetHash uploadedAtr*)r!r&r'r(r)r*r0r6reset reconnectz+live-follow: ignoring unknown event type %r)stripjsonloadsJSONDecodeErrorloggerdebugr@dictgetr/r+r7appendr rAr4r9r;)rBrCpayloadr0 raw_changesr6items rparse_follow_eventrZs&*jjl$**T" gt $ QSWY]^b_bYcd  [[*F:fc+BM MM M ")++h"7Y9OKFfc**[$2O _aefjgjaklG# !$-$Gb1b$!XXfb1:!XXf-!%)!4$3DHH\4J$K-dhhw.?@   67; ; 8O [[*Fz&#/FVQ QDQ Q  LLF MU    EtTRVSVZXs'H -H=<H=)r0sinceliveresumec^|x rdk(r \dk(r}d|iS rdk(r \dk(r}|rd|iS iS)Nr0r[r)r]r0r[s r_resume_paramsr`sE   h f% %  g U# # Irchunks should_stopc#4K|D]}|ry|yw)z Pass response chunks through until `should_stop` returns true. The check runs when a chunk arrives, so a stream carrying nothing to parse is abandoned at its next keep-alive comment instead of being read until the server ends it. Nr)rarbchunks r _iter_untilres$ =  sendpoint bucket_idtokenc#LKtjtttd}t j d|d|dit |ddit|| 5}|jtvr%td |jd |j|jjd xsd jds#t|td|jt| t!t#|j%|j'D])}t)|j*|j,}|&|+ dddy#tj.$r!} t0j3d|| Yd} ~ 8d} ~ wwxYw#1swYyxYww)aF Open a single live-follow session and yield the parsed events until the stream ends. Ends silently on EOF, on read timeout and when `should_stop` returns true. Raises [`BucketEventsUnavailable`] if the feed cannot be followed and [`~utils.HfHubHTTPError`] for anything else the caller must decide about. r)connectreadwritepoolGET /api/buckets/z/eventsrhacceptztext/event-stream)headersparamstimeoutzstatus z on z content-typerHzunexpected content-type on Nz+live-follow: stream error on bucket %s (%s))httpx2TimeoutCONNECT_TIMEOUTSTREAM_READ_TIMEOUTrstreamrr` status_code_UNFOLLOWABLE_STATUS_CODESrurlrrrU startswithrr re iter_byteseventsrZeventrC HTTPErrorrRrS) rfrgrhr]rbrtresponserparsedes r_follow_eventsrsnn_;NVelpqG    *M)G4P#%0P(>B   rcd|_yr)rrs rrLz_Backoff.reset#s  rr=N)rrrrfloatrrrLrrrrrs0JUu e rrrct|dd}||jjdnd} | t|SdS#t$rYywxYw)zAThe server `Retry-After` hint (rate limiting,...) if it sent one.rNz retry-after)getattrrrrUr ValueError)rrhints r _retry_afterr'sXq*d+H2:2F8     .DD".uT{8D8 s == A A c<t|dd}| |jSdS)Nr)rrz)rrs r _status_coder1s&q*d+H#+#78  ATArceZdZdZdddededeezdzdeeegdfdegdfd egefd dffd Z dd Z d efd Z ddZ d efdZ d efdZded dfdZxZS)ra Daemon thread keeping one bucket's event stream alive and reporting the changes it carries. One instance owns exactly one stream, for one bucket. Batches are handed over to `on_changes`. Whenever no usable resume point is known (first session, or the server cannot replay that far back), `on_reconcile` is called first: the consumer is expected to drop what it holds for that bucket, so that the replay that follows starts from a fresh state. The thread ends on its own when `is_alive` turns false, when the feed is not served, or after too many sessions ending before they were even ready. It keeps no strong reference to its consumer, which can therefore be garbage collected while the thread is running. Args: endpoint (`str`): Endpoint of the Hub. bucket_id (`str`): ID of the bucket to follow (e.g. `"username/my-bucket"`). token (`bool` or `str`, *optional*): Token used to consume the feed, resolved again at every (re)connect. on_changes (`Callable[[list[BucketFileChange]], None]`): Called with each batch of changes received from the feed. on_reconcile (`Callable[[], None]`): Called before (re)subscribing without a usable cursor. is_alive (`Callable[[], bool]`): Whether the consumer still cares about the updates. Checked before every session and whenever the stream sends anything, so a stream carrying no change is abandoned within one keep-alive. Nrprfrgrh on_changes on_reconcileis_aliver=ct|dd|||_d|_||_||_||_||_||_tj|_ d|_ y)NTzhf-bucket-live-follow-)daemonrBF) superrrg subscribed _endpoint_token _on_changes _on_reconcile _is_alive threadingEvent _stop_event_resume)rrfrgrhrrr __class__s rrzBucketFollower.__init__Qsh -CI;+OP"! %)!$??,+/ rc8|jjy)zgAsk the follower to stop. Returns before the thread ends: it stops once the stream yields or times out.N)rsetrs rstopzBucketFollower.stopgs rc\|jjxs|j S)zWWhether to give up: explicitly asked to, or the consumer holding this follower is gone.)ris_setrrs r _should_stopzBucketFollower._should_stopks&&&(@0@,@@rcttt}d}|js|j'd|j f|_|j  |j}|r(|j)d}|j!t*nO|d z }|t,k\r"tjd ||jy|j!|j'|jsyy#t$r0}tjd|jd|dYd}~yd}~wt$r}t|dk(rr|jx r%dk(r \dk(rd|j f|_n? rdk(r\dk(r d |_n( tjd |j|Yd}~yt|}|j!|t#||j$n|j'Yd}~d}~wwxYw) Nrr[zCannot live-follow bucket 'z' (z#): its listings won't be refreshed.ir_r0)r\rHz(live-follow: giving up on bucket %s (%s)r zMlive-follow: %s sessions for bucket %s ended before 'ready'; not following it)rRETRY_BASE_DELAYRETRY_MAX_DELAYrr_fetch_updated_atr _follow_oncerrRwarningrg ExceptionrrSr_sleeprrrrLRECONNECT_MIN_DELAYMAX_SESSIONS_WITHOUT_READY)rbackoffsessions_without_readyhealthyrrs rrunzBucketFollower.runos+_=!"##%||#!()?)?)AB ""$ ++-( )*& /0&!+&)-GGLLg.  GLLN+Q##%+ !X==((1$,f#5DL > VX#'DL .)(0&'9 /  " - . rc tj|jd|jt |j t }t||jjd}t|tr|SdS#t$r+}tjd|j|Yd}~yd}~wwxYw)zNGet the instant (ISO8601) to resume from. Empty string if it cannot be probed.rorp)rrrt updatedAtrHz6live-follow: cannot probe updatedAt for bucket %s (%s)N)rrUrrgrrrwrrOr@r+rrRrS)rr updated_atrs rrz BucketFollower._fetch_updated_ats "}((>>"-/?@(t{{;')H  )!,,[9J!+J!<: D" D  LLQSWSaSacd e sBB B C!B<<Crc:|jj|yr?)rwaitrs rrzBucketFollower._sleeps e$rr)rrrrr+boolrr7r rrrrrrrr __classcell__)rs@rrr6s>$( 00 0 czD 0 d#345t;< 0r4x(02t8$0 0,AdA+,Zd> 3 %E%d%r)BrrOrweakrefcollections.abcrr dataclassesrrrtypingrr r rurHr _hot_reload.sse_clientr utilsrrrr get_loggerrrRrxrwrrrrr{rrTtupler+rWeakValueDictionaryr,LockBUCKET_FOLLOWERS_LOCKrrr r/r4r9r; _FollowEventrArZ ResumePointr`bytesrerrrrr-rThreadrrrrrs> .(&& -UU   H %e2tv$uS$*t"3S895AQSZSnSnAn;oopu& (QiQ "" "<   BB B        VXvz9: O3O8d?O:S:: t0C:@G56;< ;-$sCx.  hr4x6H XV[_ (V(V(V #: (V $  (V "d(# (V l (VV"I%$,BIB#*B U%Y%%U%r