1yiUddlZddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZddlmZmZddlmZGdde ZGd d e ZGd d ZeGd deZGddeZGddZy)N)Counter) dataclass) ListUnionTupleOptional NamedTupleDictAnySequence GeneratorAnyStr) BeforeAnyAfterAny) current_timecHeZdZUeed<eed<defdZededdfdZ y)StreamEntryKeytsseqreturncT|jd|jjS)N-)rrencodeselfs Y/var/www/html/content_weaver/venv/lib/python3.12/site-packages/fakeredis/model/_stream.pyrzStreamEntryKey.encodes$''!DHH:&--// entry_keyct|tr|jn|}|jd}t |dk(rt |ddfnt |dt |df\}}t ||S)Nrr) isinstancebytesdecodesplitlenintr)r entry_key_strparts timestampsequences r parse_strzStreamEntryKey.parse_strsy3=i3OY--/U^ ##C(69%jAoU1X 2CPUVWPXM[^_def_g[hKiHi22rN) __name__ __module__ __qualname__r&__annotations__r"r staticmethodrr+rrrr s; G H003V3(833rrc0eZdZUdZeed<eed<eed<y)PelEntryzFPending Entry List entry: tracks consumer ownership and delivery count consumer_name time_readtimes_deliveredN)r,r-r.__doc__r"r/r&r1rrr3r3sPNrr3cfeZdZdZdeeeefdefdZ e de defdZ e d dededdfdZy ) StreamRangeTestz0Argument converter for sorted set LEX endpoints.value exclusivec ||_||_yN)r:r;)rr:r;s r__init__zStreamRangeTest.__init__'s "rrrcN tj|y#t$rYywxYw)NTF)rr+ ValueError)rs r valid_keyzStreamRangeTest.valid_key+s*   $ $Y /  s  $$c|dk(r|tdS|dk(r|tdS|dddk(r|tj|dddS|tj||S)N-T+r ()rrrr+)clsr:r;s rr#zStreamRangeTest.decode3sr D=y{D) ) d]xz4( ( 2AY$ ~//ab :DA A>++E2I>>rNF)r,r-r.r7rrrrboolr>r0rrA classmethodr"r#r1rrr9r9$sq:#eNIx$GH#UY#V?5?T?>O??rr9cjeZdZUeed<eed<eed<eed<deddfdZdedeee eefffd Z y) StreamConsumerInfonamepending last_attempt last_successrNcP||_d|_t}||_||_yNr)rLrMrrNrO)rrL_times rr>zStreamConsumerInfo.__init__Es'  !!r curr_timecl|j|j||jz ||jz dS)N)rLrMidleinactive)rLrMrNrO)rrSs rinfozStreamConsumerInfo.infoLs6II|| 1 11!D$5$55   r) r,r-r.r"r/r&r>r strrrWr1rrrKrK>sQ K L"U"t" c d3eSj0A+A&B rrKceZdZ d#dddededeefdZdedeed dfd Zd ed efd Z d ed efd Z d e e e eeefffdZd e eeffdZd edededed e e eee eefff dZd$dZdeed efdZdeedeedeedeedeed e e ef dZd e efdZdedeeeeefd edeed ed ee ee eff d!Zdededed e efd"Zy)% StreamGroupNstreamXStreamrL start_key entries_readct||_||_||_||_i|_||_||_i|_yr=)r[rLr]r^ consumerslast_delivered_key last_ack_keypel)rr[rLr]r^s rr>zStreamGroup.__init__VsA  "(:<"+%46rlast_delivered_strrc8|jj||_|jj|j\}}|xsd|_|jj t ||xsdzt|jdz |_y)z#Set last_delivered_id for the grouprr N) r[ parse_ts_seqr] find_indexr^ get_indexminr%ra)rrdr^ start_index_s rset_idzStreamGroup.set_idis112DE;;11$..Aa(-A"&++"7"7Kr r)rr`rKrNramaxrr+r[ stream_readr3rcr%r^rOrM format_record) rr4r{r|r}rRr]ids_readrtxs r group_readzStreamGroup.group_reads@  .,>},MDNN= )5:}%2 t //IN44X>@W@WXI;;**9e< @&}eQ?  @ x=1 &)$*A*A8B<&PD #!%!2!2!7a3x= HD 5:}%2 }%--X>-6>? ))!,???s"E4c xtj|jjdDcic]\}}|t |dj  }}}|D]V}||j vrt||j |<|||j |_|||j |_ Xycc}}w)Nc|jSr=)r4rs rz6StreamGroup._calc_consumer_last_time..s 1??r)keyc|jSr=)r5rs rrz6StreamGroup._calc_consumer_last_time..s AKKr) itertoolsgroupbyrcvaluesrir5r`rKrNrO)rrtvnew_last_success_mapconsumers r_calc_consumer_last_timez$StreamGroup._calc_consumer_last_times "))$((//*;AZ[ 1 s1/0:: :  - SHt~~-+=h+Gx(4H4RDNN8 $ 14H4RDNN8 $ 1  S  s#B6argsc:d}|D]t} tj|}||jvs(|j|j}|j |xj dzc_|j|=|dz }v|j|S#t$rYwxYwrn)rr+ Exceptionrcr4r`rMr)rrrqrtparsedr4s rackzStreamGroup.acks A '11!4! $ 0 > > }-55:5HHV$q  %%'   sB BBrUstartendrct}t|jj}|,|Dcgc]!}|j|j|k(s |#}}|/|Dcgc]$}|j|j |z|ks#|&}}|e|c|Dcgc]X}|j |ks|j |k(r8|js,|j |kDs|j |k(r|js|Z}}|t|d|}|Dcgc]^}|j|j|j||j|j z |j|jg`c}Scc}wcc}wcc}wcc}wr=) rlistrckeysr4r5r:r;sortedrr6) rrUrrr|rrR relevant_idsrts rrMzStreamGroup.pendingsnDHHMMO,  '3]!txx{7P7PT\7\A]L]  '3\!txx{7L7Lt7SV[7[A\L\  &kkAo5;;!+;EOO))a-SYY!^CMM L  !,/7L " XXZ!22EDHHQKrlrorrrr rXrrur rzrHrrrrr9rMrr rrr1rrrZrZUs( '+ 666" 6 sm 6&vvhsmvPTv%C %CPT#uUCZ/@*@%A BPD,$@"@.3@': ': } ':  ': tN#T.%99 :':RU34P^K_rrZc TeZdZdZd/dZdedeefdZdeded ee ddfd Z dede fd Z de e eeffd Zd ede efdZde ede fdZd0deeee fdededeffdZdZde fdZdee eee efedffdZdedeee effdZde defdZdedefdZd1dedede e effdZ!dede e effdZ"e#d edefd!Z$ d2d"ee d#eed$ee de fd%Z%d3d&e&d'e&d(ede efd)Z'defd*Z(d+ed,ee dfde efd-Z)dede eee eefffd.Z*y)4r\aClass representing stream. The stream contains entries with keys (timestamp, sequence) and field->value pairs. This implementation has them as a sorted list of tuples, the first value in the tuple is the key (timestamp, sequence). The structure of _values list is: [ ((timestamp, sequence), [field1, value1, field2, value2, ...]), ((timestamp, sequence), [field1, value1, field2, value2, ...]), ] rNc^g|_i|_i|_tdd|_d|_yrQ)_ids _values_dict_groupsr_max_deleted_id_entries_addedrs rr>zXStream.__init__/s/*, ?A13 -a3r group_namec:|jj|dSr=)rrrrs r group_getzXStream.group_get6s|| D11rrL start_key_strr^c|dk(r4t|jdkDr|jdn tdd}ntj|}t |||||j |<y)zAdd a group listening to stream :param name: Group name :param start_key_str: start_key in `timestamp-sequence` format, or $ listen from last. :param entries_read: Number of entries read. $rrN)r%rrr+rZr)rrLrr^r]s r group_addzXStream.group_add9s\ D ),TYY!); " PQSTAUI&00?I(tY M Trc<||jvr|j|=yy)Nr r)rrs r group_deletezXStream.group_deleteFs  % Z(rcg}|jjD]#}|j}|j|%|Sr=)rrrzr)rrqgroup group_ress r groups_infozXStream.groups_infoLsD&(\\((* "E((*I JJy ! " rfullc t|jt|jt|jdkDr|j|jdndt|jdkDr|j|jdnd|jj |j t|jdkDr|jdj ndd}|re|jDcgc]}|j|c}|d<|jjDcgc]}|jc}|d<ttj|jScc}wcc}w)Nrrs0-0)slengthgroupss first-entrys last-entrysmax-deleted-entry-ids entries-addedsrecorded-first-entry-idsentriesr) r%rrrrrrrrzrrchainitems)rrrqrgs r stream_infozXStream.stream_infoSs&499~4<<(@CDIIQR@RD..tyy|BiiHt11!4HC O6:ll6I6I6KLallnLC NIOOSYY[122ILs F>F lstcd}|D]m}|j|\}}|st|j||j|_|j|j|=|j|=|dz }o|S)zDelete items from stream :param lst: List of IDs to delete, in the form of `timestamp-sequence`. :returns: Number of items deleted rr )find_index_key_as_strrrrr)rrrqitemindfounds rdeletezXStream.deletebs}  D33D9JC'*499S>4;O;O'P$%%diin5IIcNq   rfieldsrcVt|dzdk7r tdt|tr|j }||dk(rt dt j zd}}t|jdkDrm|jdj|k\rQ|jdj|k\r5|jdj}|jdjdz}t||}n|ddk(r|jd }t|dk7ryt |d|d}}t|jdkDr9||jdjk(r|jdjdz}nd}t||}ntj|}t|jdkDr|jd|kDry|jj|t||j|<|xj dz c_|j#S) a+Add entry to a stream. If the entry_key cannot be added (because its timestamp is before the last entry, etc.), nothing is added. :param fields: List of fields to add, must [key1, value1, key2, value2, ... ] :param entry_key: Key for the entry, formatted as 'timestamp-sequence' If entry_key is '*', the timestamp will be calculated as current time and the sequence based on the last entry key of the stream. If entry_key is 'ts-*', and the timestamp is greater or equal than the last entry timestamp, then the sequence will be calculated accordingly. :returns: The key of the added entry. None if nothing was added. :raises AssertionError: If len(fields) is not even. rz The number of fields is not evenN*irr r)r%AssertionErrorr!r"r#r&timerrrrr$r+rrrrr)rrrrrts_seqr$s raddz XStream.addrs$ v;?a  !CD D i '!((*I   S 0$,-qB499~!diim&6&6"&<2ARARVYAYYYr]%%iim''!+#B,F r]c !OOC(E5zQ%(mU1XB499~!bDIIbM,<,<&<iim''!+#B,F#--i8F tyy>A $))B-&"8  $(L&! q }}rcy)NTr1rs r__bool__zXStream.__bool__src,t|jSr=)r%rrs r__len__zXStream.__len__s499~rc|dttttttftdfffd }|S)Nrc3VKjD]}j|ywr=)rr)rtrs rgenzXStream.__iter__..gens+YY ,((++ ,s&))r rrr"r )rrs` r__iter__zXStream.__iter__s7 ,YtE%e*<$=>TIJ ,u rrcF|jtj|Sr=)rrr+rrs r __getitem__zXStream.__getitem__s  !9!9#!>??rrc |j|Sr=)r)rrs rrhzXStream.get_indexsyy~rc||jvSr=)rrs r __contains__zXStream.__contains__sd''''r from_leftc$t|jdk(ry|r#tj|j|}|}n%tj|j|}|dz }||t|jkxr|j||k(fS)a^Find the closest index to entry_key_str in the stream :param entry_key: Key for the entry. :param from_left: If not found exact match, return index of last smaller element :returns: A tuple (index of entry with the closest (from the left) key to entry_key_str, whether the entry key is equal) r)rFr )r%rbisect bisect_left bisect_right)rrrr check_idxs rrgzXStream.find_indexs tyy>Q  $$TYY :CI%%dii;CaIYTYY/UDIIi4HI4UVVrr'c|dk(r$tt|jdz ddfStj |}|j |S)a,Find the closest index to entry_key_str in the stream :param entry_key_str: key for the entry, formatted as 'timestamp-sequence.' :returns: A tuple (index of entry with the closest (from the left) key to entry_key_str, whether the entry key is equal) rr rT)rr%rrr+rg)rr'rs rrzXStream.find_index_key_as_strsJ D s499~)1-t3 3))-8v&&r ts_seq_strcN|dk(r tddStj|S)Nrr)rr+)rs rrfzXStream.parse_ts_seqs(  !!Q' ''' 33r max_lengthstart_entry_keylimitcV| | tdd}|t|j|z }n||j|\}}|}t t |xsdd|xst j}|jd||j|dc}|_|D]} |j| =|S)aTrim a stream :param max_length: Max length of the resulting stream after trimming (number of last values to keep) :param start_entry_key: Min entry-key to keep, cannot be given together with max_length. :param limit: Number of entries to keep from minid. :returns: The resulting stream after trimming. :raises ValueError: When both max_length and start_entry_key are passed. Nz/Can not use both max_length and start_entry_keyr) r@r%rrrirsysmaxsizer) rrrr start_indrexactrq remove_keysrts rtrimz XStream.trims  !o&ANO O#'  !DII3I  (33ODJCIs9>1-u/C D!%4C$))CD/ TY %A!!!$ % rrstopreversecddtdtdtffd }||}||d}t||Dcgc] }j j |"}}|rt t|S|Scc}w)zReturns a range of the stream values from start to stop. :param start: Start key :param stop: Stop key :param reverse: Should the range be in reverse order? :returns: The range between start and stop elemrrct|jtryt|jtrt j Sj |j|\}}|r|jr ||rdndz }|S)Nrr r)r!r:rrr%rrgr;)rrrrrs r _find_indexz#XStream.irange.._find_indexsd$**i0$**h/499~%Y?JCIq2-JrF)rT)r9rHr&rangerrrreversed) rrrrrrstop_indrmatchess ` rirangezXStream.iranges o $ #  & tu5=B9h=WX4%%diil3XX )* *Ys%A:ct|jdkDr|jdjSdjS)Nrrz0-0)r%rrrs r last_item_keyzXStream.last_item_keys3),TYY!);tyy}##%OOrr]r|c|j|\}}|r|dz }|t|k\rgS|||zt|k\r t|n||z}|j||Sr)rgr%r)rr]r|rrend_inds rrzXStream.stream_readsi??95 5  NI D !I$} E0ASY0N#d)T]`eTeyy7++rcD|j|}|j|gSr=)rr)rrresultss rrzXStream.format_record s###C( g&&rr)rr)NNNrG)+r,r-r.r7r>r"rrZrr&rrrr r rrHrrrr rrXrrrr rrrrhrrrgrr0rfrr9r r rrr1rrr\r\!s  2E2h{.C2 Ne NE NRU N[_ Nu T$ucz"23 3 3c 3$v,3 /(5#45/#/PUVZ\aVaPb/b)Dud5k/A)B$CS$$NO@u@~tE{/J)K@S^((4(WNWtWuUXZ^U^O_W$ '6 'eCI>N '44N44%))-# SM"#}  :O?T^bcf^g4PuP,^,E#t)rs~  !ccc3+ 3Z 3z??4     ,I&IXB'B'r