KiW ddlZddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZddlmZmZddlmZddlmZGd d e ZGd d e ZGd dZeGddeZGddeZGddeZ y)N)Counter) dataclass) ListUnionTupleOptional NamedTupleDictAnySequence GeneratorAnyStr) BeforeAnyAfterAny) current_time) BaseModelcHeZdZUeed<eed<defdZededdfdZ y)StreamEntryKeytsseqreturncT|jd|jjS)N-)rrencodeselfs c/home/jay/workspace/scripts/.codegraph-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)Nrrr) 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,rrrrs; 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#r0r'r2rrr4r4sPNrr4cfeZdZdZdeeeefdefdZ 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 "rr rcN tj|y#t$rYywxYw)NTF)rr, ValueError)r s 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.decode4sr D=y{D) ) d]xz4( ( 2AY$ ~//ab :DA A>++E2I>>rNF)r-r.r/r8rrrrboolr?r1rrB classmethodr#r$r2rrr:r:%sq:#eNIx$GH#UY#V?5?T?>O??rr:cjeZdZUeed<eed<eed<eed<deddfdZdedeee eefffd Z y) StreamConsumerInfonamepending last_attempt last_successrNcP||_d|_t}||_||_yNr)rMrNrrOrP)rrM_times rr?zStreamConsumerInfo.__init__Fs'  !!r curr_timecl|j|j||jz ||jz dS)N)rMrNidleinactive)rMrNrOrP)rrTs rinfozStreamConsumerInfo.infoMs6II|| 1 11!D$5$55   r) r-r.r/r#r0r'r?r strrrXr2rrrLrL?sQ K L"U"t" c d3eSj0A+A&B rrLceZdZ 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)% StreamGroupNstreamXStreamrM start_key entries_readct||_||_||_||_i|_||_||_i|_yr>)r\rMr^r_ consumerslast_delivered_key last_ack_keypel)rr\rMr^r_s rr?zStreamGroup.__init__WsA  "(:<"+%46rlast_delivered_strrc8|jj||_|jj|j\}}|xsd|_|jj t ||xsdzt|jdz |_y)z#Set last_delivered_id for the grouprrN) r\ parse_ts_seqr^ find_indexr_ get_indexminr&rb)rrer_ start_index_s rset_idzStreamGroup.set_idjs112DE;;11$..Aa(-A"&++"7"7K)rarXr)rks rconsumers_infozStreamGroup.consumers_info~s0@DO1q!&&|~6OOOs-?c|jj|j\}}|jj|j\}}|jj|j\}}||j xsdzt |jkDr*t |j|z |j xsdz }nt |jdz |z }|jt |jt |j|jj|j |d}|S)Nrr)snames consumersspendingslast-delivered-ids entries-readslag) r\rhr^rbrcr_r&rMrardr)rrkrllast_delivered_indexlast_ack_indexlagrrs r group_infozStreamGroup.group_infos//? Q"&++"8"89P9P"Qa KK2243D3DE $++0q 1C 4D Ddkk"[0D4E4E4JKCdkk"Q&)==CYYdnn-DHH "&"9"9"@"@"B!..   rstart_idcountnoackct}||jvrt||j|<||j|_|dk(r |j}n)t t j||j}|jj||}|s!|D]}t||d|j|<t|dkDr?t |j|d|_|jxsdt|z|_ ||j|_|j|xjt|z c_|D cgc]} |jj!| c} Scc} w)N>rr)rrarLrOrbmaxrr,r\ stream_readr4rdr&r_rPrN format_record) rr5r|r}r~rSr^ids_readruxs 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>)r5rs rz6StreamGroup._calc_consumer_last_time..s 1??r)keyc|jSr>)r6rs rrz6StreamGroup._calc_consumer_last_time..s AKKr) itertoolsgroupbyrdvaluesrjr6rarLrOrP)rruvnew_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$rYwxYwro)rr, Exceptionrdr5rarNr)rrrrruparsedr5s rackzStreamGroup.acks A '11!4! $ 0 > > }-55:5HHV$q  %%'   sB BBrVstartendrct}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>) rlistrdkeysr5r6r;r<sortedrr7) rrVrrr}rrS relevant_idsrus rrNzStreamGroup.pendingsnDHHMMO,  '3]!txx{7P7PT\7\A]L]  '3\!txx{7L7Lt7SV[7[A\L\  &kkAo5;;!+;EOO))a-SYY!^CMM L  !,/7L " XXZ!22EDHHQK)rr,rrrdr6rjr&)rrrr}r^rTrurs r read_pel_msgszStreamGroup.read_pel_msgssx",,U3  N $((tQy488A;;P;P/PT_/_efjsesqtuE3t9%FU|us$BBBr>rN)r-r.r/r#rrr'r?rmrprsrr rYrrvr r{rIrrrrr:rNrr rrr2rrr[r[Vs( '+ 666" 6 sm 6&vvhsmvPTv%C %CPT#uUCZ/@*@%A BPD,$@"@.3@': ': } ':  ': tN#T.%99 :':RU34P^K_rr[c XeZdZdZdZd0dZdedeefdZ ded ed ee ddfd Z dede fd Z de eeeffd Zdede efdZde ede fdZd1deeee fdededeffdZdZde fdZdee eee efedffdZdedeee effdZde defdZdedefdZ d2dedede!e effdZ"dede!e effd Z#e$d!edefd"Z% d3d#ee d$eed%ee de fd&Z&d4d'e'd(e'd)ede efd*Z(defd+Z)d,ed-ee dfde efd.Z*dede eeeeefffd/Z+y)5r]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, ...]), ] sstreamrNclg|_i|_i|_tdd|_d|_d|_yrR)_ids _values_dict_groupsr_max_deleted_id_entries_added_last_generated_idrs rr?zXStream.__init__2s7*, ?A13 -a337r group_namec:|jj|dSr>)rrrrs r group_getzXStream.group_get:s|| D11rrM 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,r[r)rrMrr_r^s r group_addzXStream.group_add=s\ D ),TYY!); " PQSTAUI&00?I(tY M Trc<||jvr|j|=yy)Nrr)rrs r group_deletezXStream.group_deleteJs  % Z(rcg}|jjD]#}|j}|j|%|Sr>)rrr{r)rrrgroup group_ress r groups_infozXStream.groups_infoPsD&(\\((* "E((*I JJy ! " rfullc t|jt|jt|jdkDr|j|jdnd|jr |jndt|jt|jt|jdkDr|j|jdnd|j j |jt|jdkDr|jdj ndd }|re|jDcgc]}|j|c}|d<|jjDcgc]}|jc}|d<ttj|jScc}wcc}w)Nrrs0-0) slengthgroupss first-entryslast-generated-idsradix-tree-keyssradix-tree-nodess last-entrysmax-deleted-entry-ids entries-addedsrecorded-first-entry-idsentriesr)r&rrrrrrrrr{rrchainitems)rrrrrgs r stream_infozXStream.stream_infoWsP499~4<<(@CDIIQR@RD..tyy|BiiHt11!4HC O6:ll6I6I6KLallnLC NIOOSYY[122ILs G?G 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)rrrritemindfounds rdeletezXStream.deleteis}  D33D9JC'*499S>4;O;O'P$%%diin5IIcNq   rfieldsr ct|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#|_|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*irrr)r&AssertionErrorr"r#r$r'timerrrrr%r,rrrrrr)rrr rrts_seqr%s raddz XStream.addys$ 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)NTr2rs r__bool__zXStream.__bool__src,t|jSr>)r&rrs r__len__zXStream.__len__s499~rc|dttttttftdfffd }|S)Nrc3VKjD]}j|ywr>)rr)rurs 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 rrizXStream.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)rr rr check_idxs rrhzXStream.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) rrrT)rr&rrr,rh)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 rrgzXStream.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) rAr&rrrjrsysmaxsizer) rrrr start_indrexactrr remove_keysrus 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)Nrrr)r"r;rrr&rrhr<)rrrrrs r _find_indexz#XStream.irange.._find_index sd$**i0$**h/499~%Y?JCIq2-JrF)rT)r:rIr'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)rhr&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)NNNrH),r-r.r/r8 _model_typer?r#rr[rr'rrrr r rrIrrrr rrYrrrr rrrrirrrhrr1rgrr:r rrrr2rrr]r]"s K82E2h{.C2 Ne NE NRU N[_ Nu T$ucz"2333c3$$v,3 0(5#450#0PUVZ\aVaPb0d)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&IXI'iI'r