M\
mge
d dl mZ d dlZd dlZd dlZd dlZd dlmZ d dlm Z m
Z
m Z d dl Z
d dlZ
d dlmZ e rd dlmZ dZdZd Zd
Z ej e Z ej d ZddZ G d d Z G d d Z dS ) ) annotationsN) dataclass)
TYPE_CHECKINGListOptional)api)JetStreamContextz KV-OperationDELPURGEsubz^[-/_=\.a-zA-Z0-9]+$keystrreturnboolc t | dk s| d dk s | d dk rdS t t | S )Nr .F)lenr VALID_KEY_REmatchr
s j/builddir/build/BUILD/imunify360-venv-2.6.2/opt/imunify360/venv/lib/python3.11/site-packages/nats/js/kv.py
_is_key_validr ) sJ
3xx1}}A#
RC u ""3'' ( (( c e Zd ZdZdS )StopIterSentinelz=A sentinel class used to indicate that iteration should stop.N)__name__
__module__ __qualname____doc__ r r r r / s GGDr r c * e Zd ZdZe G d d Z ed G d d Zd3dZd4d5dZd6d7dZ d8d9d Z
d:d;d Z dd#Z
d6d?d$Zd@dAd'ZdBd(Z G d) d* ZdCd+Zd6dDd.ZdEd0Z dFdCd2ZdS )GKeyValuea
KeyValue uses the JetStream KeyValue functionality.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
# Create a KV
kv = await js.create_key_value(bucket='MY_KV')
# Set and retrieve a value
await kv.put('hello', b'world')
entry = await kv.get('hello')
print(f'KeyValue.Entry: key={entry.key}, value={entry.value}')
# KeyValue.Entry: key=hello, value=world
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
c Z e Zd ZU dZded< ded< ded< ded< ded < ded
< d ed < d
S )KeyValue.Entryz>
An entry from a KeyValue store in JetStream.
r bucketr
zOptional[bytes]value
Optional[int]revisiondeltacreatedz
Optional[str] operationN)r r r r __annotations__r! r r Entryr% R sl r r. T)frozenc p e Zd ZU dZded< ded< ed
d Zed
d Zedd Zd S )KeyValue.BucketStatuszB
BucketStatus is the status of a KeyValue bucket.
zapi.StreamInfo stream_infor r& r intc $ | j j j S )zU
values returns the number of stored messages in the stream.
)r2 statemessagesselfs r valuesz KeyValue.BucketStatus.valuesi s
#)2 2r c $ | j j j S )zC
history returns the max msgs per subject.
)r2 configmax_msgs_per_subjectr7 s r historyz KeyValue.BucketStatus.historyp s
#*? ?r Optional[float]c J | j j j dS | j j j S )z=
ttl returns the max age in seconds.
N)r2 r; max_ager7 s r ttlzKeyValue.BucketStatus.ttlw s'
&.6t#*2 2r N)r r3 )r r> ) r r r r r- propertyr9 r= rA r! r r BucketStatusr1 ` s $### 3 3 3
3
@ @ @
@
3 3 3
3 3 3r rC namer streamprejsr directr r Nonec L || _ || _ || _ || _ || _ d S N)_name_stream_pre_js_direct)r8 rD rE rF rG rH s r __init__zKeyValue.__init__ s+
r Nr
r) r(
validate_keysc 8 K |r%t | st j j j d} | || d{V }nP# t j j j $ r4}t j j |j |j d}~ww xY w|S )z;
get returns the latest value for the key.
N)
r natsrG errorsInvalidKeyError_getKeyDeletedErrorKeyNotFoundErrorentryop)r8 r
r) rR rZ errs r getz KeyValue.get s 1 s!3!3 1'.0 0 E ))C22222222EEw~- E E E'.11#)SVDD D E s A
B#/BBc K d }| j | } |r.| j | j || j d {V }n.| j | j ||| j d {V }n3# t
j j j $ r t
j j j w xY w||j
k r1t
j j d| d|j
d t | j
||j |j d d d }|j r[|j t$ d }|t& k s |t( k r%t
j j || |S )N)seqrH )subjectr_ rH z
expected 'z ', but got '')messager& r
r' r) r* r+ r, )rN rO get_msgrM rP rT rG rU
NotFoundErrorrY r` r# r. rL datar_ headersr] KV_OPKV_DELKV_PURGErX )r8 r
r) msgr` rZ r[ s r rW z
KeyValue._get s Y%%% 2
H , ,L < - !H , ,L # < - w~+ 2 2 2'.1 1 2 ck ! !'.11:jw:j:j\_\g:j:j:j1kk k:(W
; @ --BV||rX~~gn44UB??? s A A0 00B r' bytesr3 c K |r3t | s$t j j | | j | j | | d{V }|j S )aV
put will place the new value for the key into the store
and return the revision number.
Note: This method does not support TTL. Use create() if you need TTL support.
:param key: The key to put
:param value: The value to store
:param validate_keys: Whether to validate the key format
N) r rT rG rU rV rO publishrN r_ )r8 r
r' rR pas r putz KeyValue.put st 6 s!3!3 6'.0055 58##ty$7#$7$7??
?
?
?
?
?
?v
r msg_ttlr> c K |r3t | s$t j j | d} | ||d|| d{V }n# t j j j $ rx} | | d{V |# t j j j $ r4}| |||j j
|| d{V }Y d}~nd}~ww xY wY d}~nd}~ww xY w|S )a5
create will add the key/value pair iff it does not exist.
:param key: The key to create
:param value: The value to store
:param validate_keys: Whether to validate the key format
:param msg_ttl: Optional TTL (time-to-live) in seconds for this specific message
Nr )lastrR rq ) r rT rG rU rV updateKeyWrongLastSequenceErrorrW rX rZ r) )r8 r
r' rR rq ro r\ s r createzKeyValue.create sP 6 s!3!3 6'.0055 5
{{3A]\c{ddddddddBBw~7
iinn$$$$$$$
7>1
;;SY%7}^e '
' . s5 A C-6 BC ,*CC(C C((C-rs c K |r3t | s$t j j | i }|sd}t | |t j j < d} | j
| j | ||| d{V }nX# t j j j $ r<}|j
dk r*t j j |j |d}~ww xY w|j S )z
update will update the value if the latest revision matches.
Note: TTL parameter is accepted for internal use by create(), but should not be
used directly on update operations per NATS KV semantics.
r Nrg rq iW' ) description)r rT rG rU rV r r Header EXPECTED_LAST_SUBJECT_SEQUENCErO rn rN APIErrorerr_coderu ry r_ ) r8 r
r' rs rR rq hdrsro r\ s r rt zKeyValue.update s 6 s!3!3 6'.0055 5 D:=d))SZ
67
x''49(;c(;(;UDZa'bbbbbbbbBBw~& |u$$gn>>3?>[[[ v
s "-B C%)7C C%c @ K |r3t | s$t j j | i }t
|t < |r'|dk r!t | |t j j
< | j | j
| || d{V dS )ac
delete will place a delete marker and remove all previous revisions.
:param key: The key to delete
:param last: Expected last revision number (for optimistic concurrency)
:param validate_keys: Whether to validate the key format
:param msg_ttl: Optional TTL (time-to-live) in seconds for the delete marker
r rx NT)r rT rG rU rV ri rh r r rz r{ rO rn rN )r8 r
rs rR rq r~ s r deletezKeyValue.delete s 6 s!3!3 6'.0055 5 U HD1HH>A$iiD: ;h $) 2S 2 2D' RRRRRRRRRtr c K i }t |t < t |t j j < | j | j | || d{V dS )z
purge will remove the key and all revisions.
:param key: The key to purge
:param msg_ttl: Optional TTL (time-to-live) in seconds for the purge marker
rx NT) rj rh MSG_ROLLUP_SUBJECTr rz ROLLUPrO rn rN )r8 r
rq r~ s r purgezKeyValue.purge7 sf U "4SZ
h $) 2S 2 2D' RRRRRRRRRtr olderthanc K | d{V }g }|2 3 d{V }| n7|j t k s|j t k r| | A6 |D ]}d}| j |j }t j t j j
|j z
}|dk r|| k rd}| j
| j || d{V dS )zo
purge will remove all current delete markers older.
:param olderthan: time in seconds
Nr )r` keepT)watchallr, ri rj appendrN r
datetimenowtimezoneutcr+
total_secondsrO purge_streamrM ) r8 r watcherdelete_markersrt rZ r r` durations r
purge_deleteszKeyValue.purge_deletesD s?
'''''' # . . . . . . .&~ 6))V-=-I-I %%f--- $ $ R RED /EI//G (,,X->-BCCemSH1}}X-C-C-E-E!E!E('' gD'QQ Q Q Q Q Q Q Q Qts A"c K | j | j d{V }t || j S )zL
status retrieves the status and configuration of a bucket.
N)r2 r& )rO r2 rM r# rC rL )r8 infos r statuszKeyValue.status\ sL X))$,77777777$$dj$IIIr c B e Zd Z e Zd Zd ZddZd Zd Z dS ) KeyValue.KeyWatcherc r || _ t j d | _ d | _ d | _ d| _ d S )N )maxsizeF)rO asyncioQueue_updates_sub_pending
_init_done)r8 rG s r rQ z KeyValue.KeyWatcher.__init__f s: DHU\UbknUoUoUoDM DI+/DM $DOOOr c K | j d{V | j t j j d{V dS )z6
stop will stop this watcher.
N)r unsubscriber rp r#
KeyWatcher STOP_ITERr7 s r stopzKeyValue.KeyWatcher.stopp sb )'')) ) ) ) ) ) ) )-##H$7$ABB B B B B B B B B Br @c K t j | j | d{V S # t j $ r t
j j w xY w)zI
updates fetches the next update from a watcher.
N)r wait_forr r] TimeoutErrorrT rU )r8 timeouts r updateszKeyValue.KeyWatcher.updatesw sg
/ $ -dm.?.?.A.A7 K KKKKKKKK'
/
/
/k..
/s 16 !Ac | S rK r! r7 s r __aiter__z KeyValue.KeyWatcher.__aiter__ s Kr c K | j d {V }t |t rt |S rK )r r]
isinstancer StopAsyncIteration)r8 rZ s r __anext__z KeyValue.KeyWatcher.__anext__ sN
"m / / 1 1111111 e%566 -,, r N)r )
r r r r r rQ r r r r r! r r r r c su $$&& $ $ $ C C C / / / / r r c , K | j di | d{V S )zP
watchall returns a KeyValue watcher that matches all the keys.
>N)r )watch)r8 kwargss r r zKeyValue.watchall s2 TZ..v.........r filters List[str]c 6 K | dd d{V }g } |j d{V }|r,|r*|j j dk rt
d n# t $ r}|d}~ww xY w|2 3 d{V s nT|r6t fd|D r| j
C| j
^6 | d{V |st j
j j |S )z
Returns a list of the keys from a KeyValue store.
Optionally filters the keys based on the provided filter list.
T)ignore_deletes meta_onlyNr z/Server may ignore filters if version is < 2.10.c 3 * K | ]
}|j v V d S rK r ).0fr
s r z KeyValue.keys.. s) 55qCG|555555r )r r
consumer_infor; filter_subjectloggerwarning Exceptionanyr r
r rT rG rU NoKeysError)r8 r r r keysr er
s @r r z
KeyValue.keys s
&
"),"<"<">"> > > > > > >M
V
V '6#==NN#TUUU G ! % % % % % % %#
%5555W55555 )KK((( CG$$$$ ! llnn -'., , s A
A2 2
B<A>>BC& List[Entry]c K | |d d{V }g }|2 3 d{V }|s n| | !6 | d{V |st j j j |S )zA
history retrieves a list of the entries so far.
T)include_historyN)r r r rT rG rU r )r8 r
r entriesrZ s r r= zKeyValue.history s
3
========" " " " " " " "%
NN5 ! ! ! ! # llnn -'., ,s AFc
K j | }t t j
fd}d} |st
j j } |sd} j ||d| || d{V _
t j d d{V j
d{V }
|
j
_ j
j }
d |
j
dk r-| dk r'j d d{V d_ nG# t( $ r:}
j
d{V | d} ~ ww xY wS )z
watch will fire a callback when a key that matches the keys
pattern is updated.
The first update after starting the watch is None in case
there are no pending updates.
c K s&t j j j d {V | j }d }| j rt | j v rr| j t }rQ|t k s |t k r;|j dk r.j s'j
d d {V d_ d S t j | j t' j d | j |j j |j |j | }j
| d {V |j dk r0j s+j
d d {V d_ d S d S d S )N)r r Trc )doner r rO _timeoutmetadataheaderrh r] rj ri num_pendingr r rp r# r. rL r` r rN rf sequencerE timestamp)rk metar[ rZ r
init_setupr8 r s r
watch_updatesz%KeyValue.watch..watch_updates s ??$$
N &z48;LMMMMMMMMMM+;+;
* * * * * * * *D @ . ?N " (!' !X / / !)" 1
0
mA !,4466666666E $ 0G |-H ! !$ ' ' ' A%%(a-- &**4000000000%)" ,**,, , , , , , , ,I
s &BD+ +
E/55E**E/) rD r rE r rF r rG r rH r r rI )NT)r
r r) r( rR r r r. rK )r
r r) r( r r. )T)r
r r' rl rR r r r3 )TN)
r
r r' rl rR r rq r> r r3 )NTN) r
r r' rl rs r( rR r rq r> r r3 )
r
r rs r( rR r rq r> r r )r
r rq r> r r )r )r r3 r r )r rC )r r )r r r r )r
r r r )FFFFN)r r r r r r. rC rQ r] rW rp rv rt r r r r r r r r= r r! r r r# r# 5 s9 8 ! ! ! ! ! ! ! Y ! Yd 3 3 3 3 3 3 3 3> ( ( ( ( (T "& & & & &X # "#'
D lp . 0J J J J% % % % % % % % N/ / / / ' ' ' ' 'R 0 [ [ [ [ [ [ [r r# )r
r r r )
__future__r r r loggingre dataclassesr typingr r r nats.errorsrT nats.js.errorsnats.jsr r rh ri rj r getLoggerr r compiler r r r# r! r r r sb # " " " " " ! ! ! ! ! ! 0 0 0 0 0 0 0 0 0 0 )(((((( 8 $ $rz122 ) ) ) ) t t t t t t t t t tr