o ;iw@sddlZddlZddlZddlZddlZddlZddlmZmZddl m Z ddl m Z m Z mZmZmZmZmZmZddlmZddlmZmZmZddlmZdd lmZmZmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'm(Z(m)Z)ddl*m+Z+e r de,ddde,de#dee-ee.e,fdff ddZ/ d:de,ddde#de0dee1dee-ee.e0fdff ddZ2ede,e.Z3Gddde e3Z4dee.dfdee,dffd d!Z5dee.dfdee.dffd"d#Z6e Gd$d%d%Z7d&e7dee.dffd'd(Z8Gd)d*d*Z9Gd+d,d,Z:Gd-d.d.e e3Z;Gd/d0d0e e3ZGd4d5d5Z?Gd6d7d7Z@Gd8d9d9ZAee=ZBeeAZCdS);N)AsyncGenerator AsyncIterator) dataclass) TYPE_CHECKINGGenericLiteralOptionalTextIOTypeVarUnioncast)StreamTerminatedError) ClientClosedExecTimeoutError InvalidError)api_pb2)aclosingsynchronize_api synchronizer)TaskCommandRouterClient)_Client)logger) ConflictError InternalError ServiceError) StreamType sandbox_idfile_descriptor api_pb2.FileDescriptor.ValueType last_entry_idclientreturncCsptj||d|d}|jj|2z#3dHW}|j}|jD] }|jd|fVq|j r4d|fVdSq6dS)N7)rrtimeoutr utf-8) rSandboxGetLogsRequeststubSandboxGetLogs unary_streamentry_iditemsdataencodeeof)rrr r!req log_batchmessager2D/home/ubuntu/.local/lib/python3.10/site-packages/modal/io_streams.py_sandbox_logs_iterator%s   r4 process_id last_indexdeadlinec Cstj|d|d|d}|jj|} z|r|tnd}tj| |dIdH}Wntj y9dVYdSt yBYdSw|j D] } | j |jfVqF|dr]d|jfVdSq)Nr#T)exec_idr$r get_raw_byteslast_batch_index)r$)N exit_code)rContainerExecGetOutputRequestr'ContainerExecGetOutputr)time monotonicasynciowait_for __anext__ TimeoutErrorStopAsyncIterationr+ message_bytes batch_indexHasField) r5rr!r6r7r/stream remainingbatchitemr2r2r3 _container_process_logs_iterator8s2    rMTc@seZdZUdZeeedfed<ej dddfddde d e d d e d ed e de deeddfddZedefddZdefddZddZdeeeee fdffddZd$de deedffddZdeedffddZdeedffd d!Zd"d#ZdS)%_StreamReaderThroughServerz9A StreamReader implementation that reads from the server.N_streamTFrr object_id object_typesandboxcontainer_processr! stream_typetextby_liner7r"c Cs||_||_||_||_d|_d|_d|_||_|dkr"|s"td||_ ||_ d|_ t |t s9tdt||dkrF|t jkrFtd||_|jd kr[g|_t||_dSdS) mdmd:hiddenNrTz,Sandbox streams must have text mode enabled.Fz,stream_type must be of type StreamType, got zSandbox streams must be piped.rU)_file_descriptor _object_type _object_id_clientrP_last_entry_id _line_buffer _deadline ValueError_text_by_liner. isinstancer TypeErrortypePIPE _stream_type_container_process_bufferrA create_task!_consume_container_process_stream_consume_container_process_task) selfrrQrRr!rVrWrXr7r2r2r3__init__bs,    z#_StreamReaderThroughServer.__init__cC|jSz6Possible values are `1` for stdout and `2` for stderr.r\ror2r2r3rsz*_StreamReaderThroughServer.file_descriptorcst|jd|jd|jr._run)rArlr)rorr2rtr3rs zD_StdoutPrintingStreamReaderThroughCommandRouter._start_printing_taskc tdNz8Output can only be retrieved using the PIPE stream type.rrtr2r2r3rz4_StdoutPrintingStreamReaderThroughCommandRouter.readcCtdrrrtr2r2r3rz9_StdoutPrintingStreamReaderThroughCommandRouter.__aiter__crrrrtr2r2r3rCrz9_StdoutPrintingStreamReaderThroughCommandRouter.__anext__cs\|jdur,|jttj|jIdHWdn1s"wYd|_dSdSr)rcancel contextlibsuppressrACancelledErrorrtr2r2r3rs   z6_StdoutPrintingStreamReaderThroughCommandRouter.acloser"N)rrrrr rrrrprrrrrNrrrrCrr2r2r2r3rs       rc@sbeZdZdZdddZedefdd Zdefd d Z de efd d Z defddZ ddZ dS)_DevnullStreamReaderzStreamReader implementation for a stream configured with StreamType.DEVNULL. Throws an error if read or any other method is called. rrr"NcCs ||_dSrrs)rorr2r2r3rprz_DevnullStreamReader.__init__cCrqrrsrtr2r2r3rsz$_DevnullStreamReader.file_descriptorcr)NzEread is not supported for a stream configured with StreamType.DEVNULLrcrtr2r2r3r rz_DevnullStreamReader.readcCr)NzJ__aiter__ is not supported for a stream configured with StreamType.DEVNULLrrtr2r2r3rrz_DevnullStreamReader.__aiter__cr)NzJ__anext__ is not supported for a stream configured with StreamType.DEVNULLrrtr2r2r3rCrz_DevnullStreamReader.__anext__cr)NzGaclose is not supported for a stream configured with StreamType.DEVNULLrrtr2r2r3rrz_DevnullStreamReader.aclose)rrr"N)rrrrrprrrrNrrrrCrr2r2r2r3rs  rc@seZdZUdZeeeeee fe d<dZ e e edfe d<ejdddddfddd ed ed d ed ededede ede ede eddfddZedefddZdefddZde edffddZdefddZddZdS) _StreamReaderzRetrieve logs from a stream (`stdout` or `stderr`). As an asynchronous iterable, the object supports the `for` and `async for` statements. Just loop over the object to read in chunks. _implN _read_genTFrrrQrRrSr!rVrWrXr7rrr"c Csttjddks J|r|std| dur&t|||||||||_dS| dus,J|dks2J|tjkr>t ||_dS|tj ksJ|tj ksJJt || || |} |rZt | |} nt| } |tj krjt| |_dS| |_dS)rYF)startz/line-buffering is only supported when text=TrueNrU)rAget_running_loopr _get_looprcrOrrrrrirrrrr) rorrQrRr!rVrWrXr7rrrrr2r2r3rp(s*       z_StreamReader.__init__cCrrr)rrrtr2r2r3rVsz_StreamReader.file_descriptorcstt|jIdHS)ruN)r rNrrrtr2r2r3r[sz_StreamReader.readcCs&|jstttdf|j|_|jSr)rr rrNrrrtr2r2r3r_sz_StreamReader.__aiter__cs*|js||js J|jIdHS)aDeprecated: This exists for backwards compatibility and will be removed in a future version of Modal Only use next/anext on the return value of iter/aiter on the StreamReader object (treat streamreader as an iterable, not an iterator). N)rrrCrtr2r2r3rCds  z_StreamReader.__anext__cs&|jr|jIdHd|_dSdSr)rrrtr2r2r3ros  z_StreamReader.aclose)rrrrr rOrrrrrrrrrNrrirrrrrrrprrrrrrCrr2r2r2r3rs\     . ri c@sneZdZdZdededdeddfdd Zdefd d Z d e e e e efddfd dZdddZdddZdS)_StreamWriterThroughServerbProvides an interface to buffer and write logs to a sandbox or container process stream (`stdin`).rQrRrSr!r"NcCs*d|_||_||_||_d|_t|_dS)rYrFN)_indexr^r]r_ _is_closed bytearray_buffer)rorQrRr!r2r2r3rp|s  z#_StreamWriterThroughServer.__init__cCs|j}|jd7_|S)Nr)r)roindexr2r2r3_get_next_indexsz*_StreamWriterThroughServer._get_next_indexr,cCsv|jrtdt|ttttfr1t|tr|d}t|j t|t kr)t d|j |dSt dt|j)zWrite data to the stream but does not send it immediately. This is non-blocking and queues the data to an internal buffer. Must be used along with the `drain()` method, which flushes the buffer. $Stdin is closed. Cannot write to it.r%9Buffer size exceed limit. Call drain to flush the buffer./data argument must be a bytes-like object, not Nrrcrfrr memoryviewrr-rrMAX_BUFFER_SIZE BufferErrorextendrgrhrror,r2r2r3r{s  z _StreamWriterThroughServer.writecCs d|_dS)Close the write end of the stream after the buffered data is drained. If the process was blocked on input, it will become unblocked after `write_eof()`. This method needs to be used along with the `drain()` method, which flushes the EOF to the process. TNrrtr2r2r3 write_eofs z$_StreamWriterThroughServer.write_eofc st|j}|j|}z5|jdkr+|jjtj |j ||j |dIdHWdS|jj tj |j tj|||j ddIdHWdStyV}ztt|d}~ww)zFlush the write buffer and send data to the running process. This is a flow control method that blocks until data is sent. It returns when it is appropriate to continue writing data to the stream. rT)rrr.inputN)r1 message_indexr.)r8r)rrclearrr]r_r'SandboxStdinWriterSandboxStdinWriteRequestr^rContainerExecPutInputContainerExecPutInputRequestRuntimeInputMessagerrcr)ror,rrr2r2r3drains*     z _StreamWriterThroughServer.drainr)rrrrrrrrprrr rrrr{rrr2r2r2r3rys   rc@sXeZdZdedededdfddZdeeee efddfd d Z dd d Z dd dZ dS)!_StreamWriterThroughCommandRouterrQrrr"NcCs*||_||_||_d|_t|_d|_dS)NFr)r^_command_router_client_task_idrrr_offset)rorQrrr2r2r3rps  z*_StreamWriterThroughCommandRouter.__init__r,cCsv|jrtdt|ttttfr1t|tr|d}t|j t|t kr)t d|j |dSt dt|j)Nrr%rrrrr2r2r3r{s  z'_StreamWriterThroughCommandRouter.writecCs d|_dS)NTrrtr2r2r3rrz+_StreamWriterThroughCommandRouter.write_eofcsb|j}|js |r/t|j}|jj|j|j|j||dIdH|j|jt |7_dSdS)N)rr8offsetr,r.) rrrrexec_stdin_writerr^rrr)ror.r,r2r2r3rs   z'_StreamWriterThroughCommandRouter.drainr) rrrrrrpr rrrr{rrr2r2r2r3rs  rc @sveZdZdZ  ddededdedeedeed df d d Z d e e e e efd dfd dZdddZdddZdS) _StreamWriterrNrQrRrSr!rrr"cCsF|dur t||||_dS|dusJ|dksJt|||d|_dS)rYNrU)r)rrr)rorQrRr!rrr2r2r3rps    z_StreamWriter.__init__r,cCs|j|dS)aWrite data to the stream but does not send it immediately. This is non-blocking and queues the data to an internal buffer. Must be used along with the `drain()` method, which flushes the buffer. **Usage** ```python fixture:sandbox proc = sandbox.exec( "bash", "-c", "while read line; do echo $line; done", ) proc.stdin.write(b"foo\n") proc.stdin.write(b"bar\n") proc.stdin.write_eof() proc.stdin.drain() ``` N)rr{rr2r2r3r{sz_StreamWriter.writecCs|jdS)rN)rrrtr2r2r3rsz_StreamWriter.write_eofcs|jIdHdS)aFlush the write buffer and send data to the running process. This is a flow control method that blocks until data is sent. It returns when it is appropriate to continue writing data to the stream. **Usage** ```python notest writer.write(data) writer.drain() ``` Async usage: ```python notest writer.write(data) # not a blocking operation await writer.drain.aio() ``` N)rrrtr2r2r3rsz_StreamWriter.drain)NNr)rrrrrrrrrrpr rrrr{rrr2r2r2r3rs(   rr)DrArrrwrr?collections.abcrr dataclassesrtypingrrrrr r r r grpclib.exceptionsr modal.exceptionrrr modal_protor_utils.async_utilsrrr!_utils.task_command_router_clientrr!rconfigrrrrrrVrrrrr4rrrMrNrOryrrrrrrrrrrrr StreamReader StreamWriterr2r2r2r3s (        ""^"  "+1^G-H