+
4O|jM R t ^ RIHt ^ RIt^ RIHtHtHtHtH t H
t
H t ^ RI H
t
Ht ^ RIHt ^ RIHt ^ RIHtHt ^ RIHtHtHt ^ R IHt ^ R
I H t ^ R I H t H t H!t!H"t"H#t# ^ R I$H%t% ^ R
I&H't'H(t(H)t) Rt*]+! . R O4 t,]' d ^ RI-H.t. ^ RI/H0t0 ^ RI1H2t2 ^ RI3H4t4 ^ RI5H6t6 R R lt7 ! R R]](, 4 t8 ! R R]8](, 4 t9 ! R R]8](, 4 t: ! R R ]:](, 4 t;R# ) zAWatch changes on a collection, a database, or the entire cluster.) annotationsN)
TYPE_CHECKINGAnyGenericMappingOptionalTypeUnion) CodecOptions
_bson_to_dict)RawBSONDocument) Timestamp)_csotcommon)_AggregationCommand _CollectionAggregationCommand_DatabaseAggregationCommand)AsyncCommandCursor)validate_collation_or_none)ConnectionFailureCursorNotFoundInvalidOperationOperationFailure PyMongoError)_Op) _CollationIn
_DocumentType _PipelineF)AsyncClientSession)AsyncCollection)
AsyncDatabase)AsyncMongoClient)AsyncConnectionc V ^8 d QhRRRR/# ) excr returnbool )formats "[C:\Users\alokc\chatbot\backend\venv\Lib\site-packages\pymongo/asynchronous/change_stream.py __annotate__r+ M s
L
T
c B \ V \ \ 34 ' d R# \ V \ 4 ' dk V P f R# V P ^ 8 ;' d V P R4 ;' g+ V P ^ 8 ;' d V P \ 9 # R# )z5Return True if given a resumable change stream error.TFResumableChangeStreamError)
isinstancer r r _max_wire_versionhas_error_labelcode_RESUMABLE_GETMORE_ERRORS)r% s &r*
_resumabler4 M s #)>:;;#'(( ( ! !Q & \ \3+>+>?[+\S S##a'QQCHH8Q,Q S r, c j ] tR t^ZtRtR*R R lltR R lt]R R l4 t]R R
l4 t R R lt
R
R lt R R lt R R lt
R R ltR R ltR R ltR R ltR R lt]R R l4 t]P* R R l4 t]t]R! R" l4 t]P* R# R$ l4 tR% R& ltR' R( ltR)t R# )+AsyncChangeStreama The internal abstract base class for change stream cursors.
Should not be called directly by application developers. Use
:meth:`pymongo.asynchronous.collection.AsyncCollection.watch`,
:meth:`pymongo.asynchronous.database.AsyncDatabase.watch`, or
:meth:`pymongo.asynchronous.mongo_client.AsyncMongoClient.watch` instead.
.. versionadded:: 3.6
.. seealso:: The MongoDB documentation on `changeStreams `_.
Nc P V ^8 d QhRRRRRRRRR R
R R
R R
RRRRRRRRRRRRRR/# )r$ targetzdUnion[AsyncMongoClient[_DocumentType], AsyncDatabase[_DocumentType], AsyncCollection[_DocumentType]]pipelinezOptional[_Pipeline]
full_documentz
Optional[str] resume_afterOptional[Mapping[str, Any]]max_await_time_msz
Optional[int]
batch_size collationzOptional[_CollationIn]start_at_operation_timezOptional[Timestamp]session Optional[AsyncClientSession] start_aftercommentz
Optional[Any]full_document_before_changeshow_expanded_eventszOptional[bool]r& Noner( )r) s "r* r+ AsyncChangeStream.__annotate__f s 5: 5:
5: &5: %5: 25: )5: "5: *5: "55: . 5: 1 5: !5:" &3#5:$ -%5:&
'5:r, c Vf . p\ P ! RV4 p\ P ! RV4 \ V4 \ P ! RV4 RV n VP V n VP P P ' d> RV n VP VP P \ R7 R7 V n MWn \ P ! V4 V n W0n Wn V
R JV n VR JV n \ P ! T
;' g T4 V n WPn W`n Wpn Wn Wn Wn RV n V P P8 V n Wn R # )Nr9 r: batchSizeFT)document_class)
codec_options) r
validate_listvalidate_string_or_noner %validate_non_negative_integer_or_none_decode_customrL _orig_codec_options
type_registry _decoder_map with_optionsr _targetcopydeepcopy _pipeline_full_document _full_document_before_change_uses_start_after_uses_resume_after
_resume_token_max_await_time_ms _batch_size
_collation_start_at_operation_time_session_comment_closed_timeout_show_expanded_events)selfr8 r9 r: r; r= r> r? r@ rA rC rD rE rF s &&&&&&&&&&&&&&r* __init__AsyncChangeStream.__init__f sD ( H''
H=&&
F"9-44[*M #@F@T@T - - : : :"&D ".. $ 2 2 ? ? ? _ / DL "L x0 +,G)!,D!8 ".d": !]];+F+F, G"3 %#(?%
--
%9"r, c V ^8 d QhRR/# r$ r& rG r( )r) s "r* r+ rH s 3 3$ 3r, c L " V P 4 G R j xL
V n R # L 5iN)_create_cursor_cursorrg s &r* _initialize_cursor$AsyncChangeStream._initialize_cursor s ! 0 0 22 2s $" $c V ^8 d QhRR/# )r$ r& zType[_AggregationCommand]r( )r) s "r* r+ rH s " ",E "r, c
\ h)z)The aggregation command class to be used.NotImplementedErrorrp s &r* _aggregation_command_class,AsyncChangeStream._aggregation_command_class s
"!r, c V ^8 d QhRR/# )r$ r& r! r( )r) s "r* r+ rH s " ") "r, c
\ h)zZThe client against which the aggregation commands for
this AsyncChangeStream will be run.
ru rp s &r* _clientAsyncChangeStream._client s
"!r, c V ^8 d QhRR/# r$ r& zdict[str, Any]r( )r) s "r* r+ rH s r, c
R / pV P e V P VR& V P e V P VR& V P pVe V P ' d W!R& M"W!R& M V P e V P VR& V P
' d V P
VR& V# )z=Return the options dict for the $changeStream pipeline stage. fullDocumentfullDocumentBeforeChange
startAfter resumeAfterstartAtOperationTimeshowExpandedEvents)rY rZ resume_tokenr[ ra rf )rg optionsr s & r* _change_stream_options(AsyncChangeStream._change_stream_options s "$ *&*&9&9GN # , , 8262S2SG. /(( #%%%(4 %)5
&
*
*
6.2.K.KG* + % % %,0,F,FG( )r, c V ^8 d QhRR/# r~ r( )r) s "r* r+ rH s . r, c
~ / pV P e V P VR& V P e V P VR& V# )z4Return the options dict for the aggregation command.maxAwaitTimeMSrJ )r^ r_ )rg r s & r* _command_options"AsyncChangeStream._command_options sE " " .(,(?(?G$ % '#'#3#3GK r, c V ^8 d QhRR/# )r$ r& zlist[dict[str, Any]]r( )r) s "r* r+ rH s '; r, c
f V P 4 pRV/.pVP V P 4 V# )z@Return the full aggregation pipeline for this AsyncChangeStream.z
$changeStream)r extendrX )rg r
full_pipelines & r* _aggregation_pipeline'AsyncChangeStream._aggregation_pipeline s5 --/0?/I.J
T^^, r, c $ V ^8 d QhRRRRRR/# )r$ resultzMapping[str, Any]connr" r& rG r( )r) s "r* r+ rH s# &7 SW r, c
t VR, R, ' g RVR, 9 d VR, R, V n R# V P fm V P RJ d[ V P RJ dI VP ^8 d6 VP R4 V n V P f \
RV: 24 hR# R# R# R# R# R# )a% Callback that caches the postBatchResumeToken or
startAtOperationTime from a changeStream aggregate command response
containing an empty batch of change documents.
This is implemented as a callback because we need access to the wire
version in order to determine whether to cache this value.
cursor
firstBatchpostBatchResumeTokenNF
operationTimez?Expected field 'operationTime' missing from command response : )r] ra r\ r[ max_wire_versiongetr )rg r r s &&&r* _process_result!AsyncChangeStream._process_result s h --%)99%+H%56L%M"--5++u4**e3))Q.06
?0K-008*&&,Z1 9 / 4 5 6 .r, c V ^8 d QhRRRR/# )r$ rA rB r& r r( )r) s "r* r+ rH s
3
r, c
h " V P V P \ V P 4 V P 4 V P
V P R7 pV P P VP V P P V4 V\ P R7 G Rj xL
# L5i)znRun the full aggregation pipeline for this AsyncChangeStream and return
the corresponding AsyncCommandCursor.
)result_processorrD ) operationN)
rw rU r r r r rc r{ _retryable_read
get_cursor_read_preference_forr AGGREGATE)rg rA cmds && r* _run_aggregation_cmd&AsyncChangeStream._run_aggregation_cmd s -- LL & & ( ! ! # ! 1 1MM
.
\\11 NN LL - -g 6 mm 2
s B)B2+B0,B2c V ^8 d QhRR/# )r$ r& r r( )r) s "r* r+ rH s > >&8 >r, c " V P P V P 4 ;_uu_4 GR j xL
pV P VR7 G R j xL
uuR R R 4 GR j xL
# L0 L L + GR j xL
' g i R # ; i5i)N)rA )r{ _tmp_sessionrb r )rg ss & r* rn AsyncChangeStream._create_cursor sS <<,,T]];;;q 2212== <;;= <;;;s\ 2BA%BA+A'A+
B A) B'A+)B+B 1A42
B =B ? Bc V ^8 d QhRR/# rk r( )r) s "r* r+ rH s 3 3t 3r, c
" V P P 4 G Rj xL
V P 4 G Rj xL
V n R# L# \ d L.i ; i L 5i)z7Reestablish this change stream after a resumable error.N)ro closer rn rp s &r* _resumeAsyncChangeStream._resume sO ,,$$& & & " 0 0 22
' 2sA A A AA AA AA AAAAc V ^8 d QhRR/# rk r( )r) s "r* r+ rH s # #T #r, c
d " RV n V P P 4 G Rj xL
R# L5i)z Close this AsyncChangeStream.TN)rd ro r rp s &r* r AsyncChangeStream.close s" ll """s %0.0c V ^8 d QhRR/# r$ r& z AsyncChangeStream[_DocumentType]r( )r) s "r* r+ rH s ; r, c V # rm r( rp s &r* __aiter__AsyncChangeStream.__aiter__ s r, c V ^8 d QhRR/# )r$ r& r< r( )r) s "r* r+ rH s 1 1 9 1r, c
B \ P ! V P 4 # )ztThe cached resume token that will be used to resume after the most
recently returned change.
.. versionadded:: 3.9
)rV rW r] rp s &r* r AsyncChangeStream.resume_token s }}T / /00r, c V ^8 d QhRR/# )r$ r& r r( )r) s "r* r+ rH ! s %! %!M %!r, c
~ " V P ' d! V P 4 G Rj xL
pVf K0 V# \ h L5i)a` Advance the cursor.
This method blocks until the next change document is returned or an
unrecoverable error is raised. This method is used when iterating over
all changes in the cursor. For example::
try:
resume_token = None
pipeline = [{'$match': {'operationType': 'insert'}}]
async with await db.collection.watch(pipeline) as stream:
async for insert_change in stream:
print(insert_change)
resume_token = stream.resume_token
except pymongo.errors.PyMongoError:
# The AsyncChangeStream encountered an unrecoverable error or the
# resume attempt failed to recreate the cursor.
if resume_token is None:
# There is no usable resume token because there was a
# failure during AsyncChangeStream initialization.
logging.error('...')
else:
# Use the interrupted AsyncChangeStream's resume token to create
# a new AsyncChangeStream. The new stream will continue from the
# last seen insert change without missing any events.
async with await db.collection.watch(
pipeline, resume_after=resume_token) as stream:
async for insert_change in stream:
print(insert_change)
Raises :exc:`StopIteration` if this AsyncChangeStream is closed.
N)alivetry_nextStopAsyncIteration)rg docs & r* nextAsyncChangeStream.next s4 B jjj
'C
(s ==;= =c V ^8 d QhRR/# )r$ r& r' r( )r) s "r* r+ rH K s t r, c
$ V P ' * # )zDoes this cursor have the potential to return more data?
.. note:: Even if :attr:`alive` is ``True``, :meth:`next` can raise
:exc:`StopIteration` and :meth:`try_next` can return ``None``.
.. versionadded:: 3.8
)rd rp s &r* r AsyncChangeStream.aliveJ s << r, c V ^8 d QhRR/# )r$ r& zOptional[_DocumentType]r( )r) s "r* r+ rH V s ] ] 7 ]r, c
( " V P ' g5 V P P ' g V P 4 G Rj xL
V P P R4 G Rj xL
pT P P ' g RT n Tf= T P P e# T P P T n RT n T# TR, pT P P 4 ' g3 T P P ' d T P P pRT n RT n Y0n RT n T P$ ' d! \' TP( T P* 4 # T# EL8 EL \
d[ p\
T4 ' g h T P 4 G Rj xL
T P P R4 G Rj xL
p Rp?ELtRp?ii ; i \
dC p\
T4 ' g, TP ' g T P 4 G Rj xL
h Rp?i\ d T P 4 G Rj xL
h i ; i \ d' T P 4 G Rj xL
\ R4 Rhi ; i5i)a Advance the cursor without blocking indefinitely.
This method returns the next change document without waiting
indefinitely for the next change. For example::
async with await db.collection.watch() as stream:
while stream.alive:
change = await stream.try_next()
# Note that the AsyncChangeStream's resume token may be updated
# even when no changes are returned.
print("Current resume token: %r" % (stream.resume_token,))
if change is not None:
print("Change document: %r" % (change,))
continue
# We end up here when there are no recent changes.
# Sleep for a while before trying again to avoid flooding
# the server with getMore requests when no changes are
# available.
asyncio.sleep(10)
If no change document is cached locally then this method runs a single
getMore command. If the getMore yields any documents, the next
document is returned, otherwise, if the getMore returns no documents
(because there have been no changes) then ``None`` is returned.
:return: The next change document or ``None`` when no document is available
after running a single getMore or when the cursor is closed.
.. versionadded:: 3.8
NTF_idzECannot provide resume functionality when the resume token is missing.)rd ro r r _try_nextr r4 timeoutr
BaseException_post_batch_resume_tokenr] ra KeyErrorr _has_nextr[ r\ rP r rawrQ )rg changer% r s & r* r AsyncChangeStream.try_nextU s @ |||DLL$6$6$6,,.
= #|| 5 5d ;;" ||!!! DL >
||44@%)\\%J%J"04-M !%=L ||%%''DLL,Q,Q,Q <<@@L "' "& *(,% T-E-EF F
{
! <
=!#lln$$ #|| 5 5e <<<
=
c??3;;;jjl"" **, , **, "W
s .JJE<J F )E?*F .J
AJ I J;JA&J?F G'
%G"2F53#G"GG" G* J"G''G* * I5"H2H2+H.,H22 I?IIIJ J<I?=JJc V ^8 d QhRR/# r r( )r) s "r* r+ rH s "B r, c " V # 5irm r( rp s &r*
__aenter__ AsyncChangeStream.__aenter__ s
s c ( V ^8 d QhRRRRRRRR/# )r$ exc_typer exc_valexc_tbr& rG r( )r) s "r* r+ rH s( c 3 4 r, c B " V P 4 G R j xL
R # L5irm )r )rg r r r s &&&&r* __aexit__AsyncChangeStream.__aexit__ s jjls )r_ rd r` rc ro rP rY rZ r^ rQ rX r] rb rf ra rU re r\ r[ )NNN) __name__
__module__ __qualname____firstlineno____doc__rh rq propertyrw r{ r r r r r rn r r r r r applyr __anext__r r r r __static_attributes__r( r, r* r6 r6 Z s 5:n3 " " " " 0 2
*>3#
1 1 [[%! %!N I
[[] ]~ r, r6 c R ] tR tRt$ RtR]R& ]R R l4 t]R R l4 tR t R
# ) AsyncCollectionChangeStreami zA change stream that watches changes on a single collection.
Should not be called directly by application developers. Use
helper method :meth:`pymongo.asynchronous.collection.AsyncCollection.watch` instead.
.. versionadded:: 3.7
z AsyncCollection[_DocumentType]rU c V ^8 d QhRR/# )r$ r& z#Type[_CollectionAggregationCommand]r( )r) s "r* r+ (AsyncCollectionChangeStream.__annotate__ s - -,O -r, c \ # rm )r rp s &r* rw 6AsyncCollectionChangeStream._aggregation_command_class s ,,r, c V ^8 d QhRR/# r$ r& z AsyncMongoClient[_DocumentType]r( )r) s "r* r+ r s , ,8 ,r, c B V P P P # rm )rU databaseclientrp s &r* r{ #AsyncCollectionChangeStream._client s ||$$+++r, r( N
r r r r r __annotations__r rw r{ r r( r, r* r r s5 ,+
- - , ,r, r c R ] tR tRt$ RtR]R& ]R R l4 t]R R l4 tR t R
# ) AsyncDatabaseChangeStreami zA change stream that watches changes on all collections in a database.
Should not be called directly by application developers. Use
helper method :meth:`pymongo.asynchronous.database.AsyncDatabase.watch` instead.
.. versionadded:: 3.7
z AsyncDatabase[_DocumentType]rU c V ^8 d QhRR/# )r$ r& z!Type[_DatabaseAggregationCommand]r( )r) s "r* r+ &AsyncDatabaseChangeStream.__annotate__ s + +,M +r, c \ # rm )r rp s &r* rw 4AsyncDatabaseChangeStream._aggregation_command_class s **r, c V ^8 d QhRR/# r r( )r) s "r* r+ r s # #8 #r, c . V P P # rm )rU r rp s &r* r{ !AsyncDatabaseChangeStream._client s ||"""r, r( Nr r( r, r* r r s5 *)
+ + # #r, r c 2 a ] tR tRtRtR V 3R lltRtV ;t# )AsyncClusterChangeStreami zA change stream that watches changes on all collections in the cluster.
Should not be called directly by application developers. Use
helper method :meth:`pymongo.asynchronous.mongo_client.AsyncMongoClient.watch` instead.
.. versionadded:: 3.7
c V ^8 d QhRR/# r~ r( )r) s "r* r+ %AsyncClusterChangeStream.__annotate__ s r, c . < \ SV ` 4 pR VR& V# )TallChangesForCluster)superr )rg r __class__s & r* r /AsyncClusterChangeStream._change_stream_options s '02*.&'r, r( )r r r r r r r
__classcell__)r
s @r* r r s r, r ) Y [ i i)# i{' iP- iR- i{4 i|4 ? iL4 )r) s H " N N N , ) $ !
C 8 # B B & . F?;B9
_
. _D ,"3M"B ,(# 1- @ #( 8 G r,