o ;ik@sddlZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z m Z m Z ddlmZddlmZddlmZdd lmZdd lmZd d lmZd d lmZmZmZmZmZd dl m!Z!d dl"m#Z#m$Z$d dl%m&Z&m'Z'm(Z(d dl)m*Z*m+Z+d dl,m-Z-d dl.m/Z/d dl0m1Z1m2Z2d dl3m4Z4d dl5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;eGdddZGdddeddZ?e'e?Z@dS)N)AsyncGenerator AsyncIterator) dataclass)datetime)AnyOptionalUnion)Message)Status) classproperty)asynccontextmanager)api_pb2) LoadContext) EPHEMERAL_OBJECT_HEARTBEAT_SLEEP_get_environment_name_Object live_methodlive_method_gen)Resolver) deserialize serialize) TaskContextsynchronize_api!warn_if_generator_is_not_consumed)deprecation_warningwarn_if_passing_namespace)Retry)check_object_name) as_timestamptimestamp_to_localized_dt)_Client)AlreadyExistsErrorError InvalidError NotFoundErrorRequestSizeErrorResourceExhaustedErrorc@s2eZdZUdZeeed<eed<eeed<dS) QueueInfoz#Information about the Queue object.name created_at created_byN)__name__ __module__ __qualname____doc__rstr__annotations__rr2r2?/home/ubuntu/.local/lib/python3.10/site-packages/modal/queue.pyr(%s  r(c @seZdZdZedddddededeedeed df d d Z eddd dd dee dee e efdedeed e jdf ddZedddddededeedeefddZdS) _QueueManagerz8Namespace with methods for managing named Queue objects.FN)allow_existingenvironment_nameclientr)r5r6r7returncs~t|d|durtIdHn|}|rtjntj}tj|t||d}z |j |IdHWdSt y>|s;YdSw)a$Create a new Queue object. **Examples:** ```python notest modal.Queue.objects.create("my-queue") ``` Queues will be created in the active environment, or another one can be specified: ```python notest modal.Queue.objects.create("my-queue", environment_name="dev") ``` By default, an error will be raised if the Queue already exists, but passing `allow_existing=True` will make the creation attempt a no-op in this case. ```python notest modal.Queue.objects.create("my-queue", allow_existing=True) ``` Note that this method does not return a local instance of the Queue. You can use `modal.Queue.from_name` to perform a lookup after creation. Added in v1.1.2. QueueNdeployment_namer6object_creation_type) rr!from_envr &OBJECT_CREATION_TYPE_CREATE_IF_MISSING*OBJECT_CREATION_TYPE_CREATE_FAIL_IF_EXISTSQueueGetOrCreateRequestrstubQueueGetOrCreater")r)r5r6r7r<reqr2r2r3create4s& # z_QueueManager.create) max_objectscreated_beforer6r7rFrG_Queuecsdur tIdHndurdkrtdgdtdtffdd }|t|IdH} |r7n |d jjjIdH}q4fd d D}durX|dS|S) aReturn a list of hydrated Queue objects. **Examples:** ```python queues = modal.Queue.objects.list() print([q.name for q in queues]) ``` Queues will be retreived from the active environment, or another one can be specified: ```python notest dev_queues = modal.Queue.objects.list(environment_name="dev") ``` By default, all named Queues are returned, newest to oldest. It's also possible to limit the number of results and to filter by creation date: ```python queues = modal.Queue.objects.list(max_objects=10, created_before="2025-01-01") ``` Added in v1.1.2. Nrzmax_objects cannot be negativerGr8csdurdntdt}tj||d}tjt|d}j|IdH}|j t|j |kp?duo?tk}|S)Nd)rFrG)r6 pagination) minlenr ListPaginationQueueListRequestrrA QueueListextendqueues)rG max_page_sizerJrCrespfinishedr7r6itemsrFr2r3 retrieve_pages "z)_QueueManager.list..retrieve_pageTc s.g|]}tj|j|jdt|jdqS)T)is_another_apprep)rH _new_hydratedqueue_idmetadata_reprr)).0itemr7r6r2r3 s z&_QueueManager.list..) r!r=r$floatboolrr] creation_infor*)rFrGr6r7rWrTrQr2rUr3listis !   z_QueueManager.list) allow_missingr6r7rgcsbztj||d|IdH}Wn ty|sYdSwtj|jd}|jj |IdHdS)a+Delete a named Queue. Warning: This deletes an *entire Queue*, not just a specific entry or partition. Deletion is irreversible and will affect any Apps currently using the Queue. **Examples:** ```python notest await modal.Queue.objects.delete("my-queue") ``` Queues will be deleted from the active environment, or another one can be specified: ```python notest await modal.Queue.objects.delete("my-queue", environment_name="dev") ``` Added in v1.1.2. )r6Nr\) rH from_namehydrater%r QueueDeleteRequest object_id_clientrA QueueDelete)r)rgr6r7objrCr2r2r3deletes z_QueueManager.delete)r,r-r.r/ staticmethodr0rdrr!rDintrrbuiltinsrfrpr2r2r2r3r41s\ 4 Cr4c@sVeZdZUdZdZeejed<ddZ e de fddZ e deefd d Zd eefd d ZdejfddZedeedefddZeeddefdeddeedeedededf ddZeddddddedeededeeddf ddZe dNdedeeddfd d!Z eddd"dedeedeefd#d$Z!e"de#fd%d&Z$deed'e%de&e'fd(d)Z(deed*eed'e%de&e'fd+d,Z)e"ddd-deed.eddfd/d0Z*e" dOdd2d3ed*eedeedee'fd4d5Z+e" dOdd2d'e%d3ed*eedeede&e'f d6d7Z,e" 1 dOdd8d9d:e'd3ed*eedeed;e%ddf de&e'd3ed*eedeed;e%ddf d?d@Z. dNdeed;e%d>e&e'd*eefdAdBZ/deed;e%d>e&e'fdCdDZ0e"dddEdeedFede%fdGdHZ1e2e3ddIdJdeedKede4e'dffdLdMZ5dS)PrHa Distributed, FIFO queue for data flow in Modal apps. The queue can contain any object serializable by `cloudpickle`, including Modal objects. By default, the `Queue` object acts as a single FIFO queue which supports puts and gets (blocking and non-blocking). **Usage** ```python from modal import Queue # Create an ephemeral queue which is anonymous and garbage collected with Queue.ephemeral() as my_queue: # Putting values my_queue.put("some value") my_queue.put(123) # Getting values assert my_queue.get() == "some value" assert my_queue.get() == 123 # Using partitions my_queue.put(0) my_queue.put(1, partition="foo") my_queue.put(2, partition="bar") # Default and "foo" partition are ignored by the get operation. assert my_queue.get(partition="bar") == 2 # Set custom 10s expiration time on "foo" partition. my_queue.put(3, partition="foo", partition_ttl=10) # (beta feature) Iterate through items in place (read immutably) my_queue.put(1) assert [v for v in my_queue.iterate()] == [0, 1] # You can also create persistent queues that can be used across apps queue = Queue.from_name("my-persisted-queue", create_if_missing=True) queue.put(42) assert queue.get() == 42 ``` For more examples, see the [guide](https://modal.com/docs/guide/dicts-and-queues#modal-queues). **Queue partitions (beta)** Specifying partition keys gives access to other independent FIFO partitions within the same `Queue` object. Across any two partitions, puts and gets are completely independent. For example, a put in one partition does not affect a get in any other partition. When no partition key is specified (by default), puts and gets will operate on a default partition. This default partition is also isolated from all other partitions. Please see the Usage section below for an example using partitions. **Lifetime of a queue and its partitions** By default, each partition is cleared 24 hours after the last `put` operation. A lower TTL can be specified by the `partition_ttl` argument in the `put` or `put_many` methods. Each partition's expiry is handled independently. As such, `Queue`s are best used for communication between active functions and not relied on for persistent storage. On app completion or after stopping an app any associated `Queue` objects are cleaned up. All its partitions will be cleared. **Limits** A single `Queue` can contain up to 100,000 partitions, each with up to 5,000 items. Each item can be up to 1 MiB. Partition keys must be non-empty and must not exceed 64 bytes. N _metadatacCstd)z mdmd:hiddenzYQueue() is not allowed. Please use `Queue.from_name(...)` or `Queue.ephemeral()` instead.) RuntimeErrorselfr2r2r3__init__!sz_Queue.__init__r8cCstSN)r4)clsr2r2r3objects%sz_Queue.objectscCs|jSry)_namervr2r2r3r))sz _Queue.namer]cCs*|rt|tjs J||_|j|_dSdSry) isinstancer QueueMetadatartr)r|)rwr]r2r2r3_hydrate_metadata-s  z_Queue._hydrate_metadatacCs|jsJ|jSry)rtrvr2r2r3 _get_metadata3s z_Queue._get_metadata partitioncCs>|dur|d}t|dkst|dkrtd|Sd}|S)Nzutf-8r@z8Queue partition key must be between 1 and 64 characters.)encoderLr$)r partition_keyr2r2r3validate_partition_key7s z_Queue.validate_partition_keyrzr7r6_heartbeat_sleepc sdur tIdHtjtjt|djIdH}t4IdH+}tj |j d|j fdd|d|j |j |j ddVWdIdHdS1IdHsWwYdS) a=Creates a new ephemeral queue within a context manager: Usage: ```python from modal import Queue with Queue.ephemeral() as q: q.put(123) ``` ```python notest async with Queue.ephemeral() as q: await q.put.aio(123) ``` N)r<r6rhcs jSry)rAQueueHeartbeatr2r7requestr2r3bs z"_Queue.ephemeral..)sleepT)rY)r!r=r r@OBJECT_CREATION_TYPE_EPHEMERALrrArBrQueueHeartbeatRequestr\ infinite_loopr[r])rzr7r6rresponsetcr2rr3 ephemeralBs.z_Queue.ephemeralF) namespacer6create_if_missingr7r)rc sbtdt|ddtdtdtdttffdd }t|}tj||d d t||d d S) aPReference a named Queue, creating if necessary. This is a lazy method the defers hydrating the local object with metadata from Modal servers until the first time it is actually used. ```python q = modal.Queue.from_name("my-queue", create_if_missing=True) q.put(123) ``` r9zmodal.Queue.from_namerwresolver load_contextexisting_object_idcsJtj|jr tjndd}|jj|IdH}||j|j|j dS)Nr:) r r@r6r>r7rArB_hydrater\r]rwrrrrCrrr)r2r3_load|s z_Queue.from_name.._loadTr6r7)rYhydrate_lazilyr)load_context_overrides) rrrHrrrr0r^ _from_loader)r)rr6rr7rrZr2rr3ries  $  z_Queue.from_namer\c sHdtdtdtdttffdd }dd}tj||d d t|d d S) aConstruct a Queue from an id and look up the Queue metadata. This is a lazy method that defers hydrating the local object with metadata from Modal servers until the first time it is actually used. The ID of a Queue object can be accessed using `.object_id`. **Example:** ```python notest @app.function() def my_consumer(queue_id: str): queue = modal.Queue.from_id(queue_id) queue.put("Hello from remote function!") with modal.Queue.ephemeral() as q: # Pass the queue ID to a remote function my_consumer.remote(q.object_id) print(q.get()) # "Hello from remote function!" ``` rwrrrcs:tjd}|jj|IdH}||j|j|jdS)Nrh)r QueueGetByIdRequestr7rA QueueGetByIdrr\r]rrhr2r3rs z_Queue.from_id.._loadzQueue.from_id()T)r7)rYrr)rHrrrr0r)r\r7rrZr2rhr3from_ids" z_Queue.from_idracs(tddtjj|||dIdHdS)aKmdmd:hidden Delete a named Queue. Warning: This deletes an *entire Queue*, not just a specific entry or partition. Deletion is irreversible and will affect any Apps currently using the Queue. DEPRECATED: This method is deprecated; we recommend using `modal.Queue.objects.delete` instead. )iz\`modal.Queue.delete` is deprecated; we recommend using `modal.Queue.objects.delete` instead.rN)rrHr{rp)r)r7r6r2r2r3rps  z _Queue.deletecs0|}|j}t|jp dt|j|jpddS)z*Return information about the Queue object.N)r)r*r+)rrer(r)r r*r+)rwr]rer2r2r3infosz _Queue.infon_valuescsNtjj|d|d}jj|IdH}|jr%fdd|jDSgS)Nrr\rtimeoutrcg|]}t|jqSr2rrmr_valuervr2r3rbz+_Queue._get_nonblocking..)r QueueGetRequestrlrrmrAQueueGetvalues)rwrrrrr2rvr3_get_nonblockingsz_Queue._get_nonblockingrcs|dur t|}nd} d}|durt||t}tjj|||d}jj|IdH}|j rBfdd|j DS|durQt|krQ t q)NTgI@rcrr2rrrvr2r3rbrz(_Queue._get_blocking..) timerKr rrlrrmrArrqueueEmpty)rwrrrdeadlinerequest_timeoutrrr2rvr3 _get_blockings*z_Queue._get_blocking)rallrcsB|r |r tdtj|j|||d}|jj|IdHdS)a&Clear the contents of a single partition or all partitions. Warning: this is a destructive operation and will irrevocably delete data. **Examples:** ```python q = modal.Queue.from_name("my-queue", create_if_missing=True) q.clear() ``` z4Partition must be null when requesting to clear all.)r\rall_partitionsN)r$r QueueClearRequestrlrrmrA QueueClear)rwrrrr2r2r3clears z _Queue.clearT)rblockcsP|r|||dIdH}n|durtd||dIdH}|r&|dSdS)aRemove and return the next object in the queue. If `block` is `True` (the default) and the queue is empty, `get` will wait indefinitely for an object, or until `timeout` if specified. Raises a native `queue.Empty` exception if the `timeout` is reached. If `block` is `False`, `get` returns `None` immediately if the queue is empty. The `timeout` is ignored in this case. rN(Timeout is ignored for non-blocking get.rrwarningswarnr)rwrrrrr2r2r3gets z _Queue.getcs>|r ||||IdHS|durtd|||IdHS)aRemove and return up to `n_values` objects from the queue. If there are fewer than `n_values` items in the queue, return all of them. If `block` is `True` (the default) and the queue is empty, `get` will wait indefinitely for at least 1 object to be present, or until `timeout` if specified. Raises the stdlib's `queue.Empty` exception if the `timeout` is reached. If `block` is `False`, `get` returns `None` immediately if the queue is empty. The `timeout` is ignored in this case. Nrr)rwrrrrr2r2r3get_many0s  z_Queue.get_manyiQr partition_ttlvrcs"|j|g||||dIdHdS)aAdd an object to the end of the queue. If `block` is `True` and the queue is full, this method will retry indefinitely or until `timeout` if specified. Raises the stdlib's `queue.Full` exception if the `timeout` is reached. If blocking it is not recommended to omit the `timeout`, as the operation could wait indefinitely. If `block` is `False`, this method raises `queue.Full` immediately if the queue is full. The `timeout` is ignored in this case.rN)put_many)rwrrrrrr2r2r3putGs z _Queue.putvscsJ|r|||||IdHdS|durtd||||IdHdS)aAdd several objects to the end of the queue. If `block` is `True` and the queue is full, this method will retry indefinitely or until `timeout` if specified. Raises the stdlib's `queue.Full` exception if the `timeout` is reached. If blocking it is not recommended to omit the `timeout`, as the operation could wait indefinitely. If `block` is `False`, this method raises `queue.Full` immediately if the queue is full. The `timeout` is ignored in this case. Nz3`timeout` argument is ignored for non-blocking put.)_put_many_blockingrr_put_many_nonblocking)rwrrrrrr2r2r3r[s  z_Queue.put_manyc sdd|D}tj|j||||d}z|jjj|ttj gdd|ddIdHWdSt y_}z'dt |vrMt |dkrBd nd }t d |d |t|trYtt ||d}~ww) NcSg|]}t|qSr2rr_rr2r2r3rbxz-_Queue._put_many_blocking..r\rrpartition_ttl_seconds>@)additional_status_codes max_delay max_retries total_timeout)retrystatus = '413'rrrQueue. request is too large)r QueuePutRequestrlrrmrAQueuePutrr RESOURCE_EXHAUSTEDr#r0rLr&r}r'rFull) rwrrrr vs_encodedrexcmethodr2r2r3rus6  z_Queue._put_many_blockingc sdd|D}tj|j||||d}z |jj|IdHWdStyU}z'dt|vrCt |dkr8dnd}t d|d |t |t rOt t||d}~ww) NcSrr2rrr2r2r3rbrz0_Queue._put_many_nonblocking..rrrrrrr)r rrlrrmrArr#r0rLr&r}r'rr)rwrrrrrrrr2r2r3rs&  z_Queue._put_many_nonblocking)rtotalrcsD|r |r tdtj|j|||d}|jj|IdH}|jS)z4Return the number of objects in the queue partition.z4Partition must be null when requesting total length.)r\rrN) r$r QueueLenRequestrlrrmrAQueueLenrL)rwrrrrr2r2r3rLsz _Queue.len)ritem_poll_timeoutrc Csd}||}t|}d} tdt||t}tj|j|||d}|jj |IdH} | j rL| j D] } t | j |jV| j }q7t|}nt|krTdSq)z(Beta feature) Iterate through items in the queue without mutation. Specify `item_poll_timeout` to control how long the iterator should wait for the next time before giving up. NrTr)r\r last_entry_idr)rrmaxrKr QueueNextItemsRequestrlrmrAQueueNextItemsrVrrentry_id) rwrrrvalidated_partition_keyfetch_deadlineMAX_POLL_DURATION poll_durationrrr`r2r2r3iterates,    z_Queue.iteratery)TN)6r,r-r.r/rtrr r~r1rxr r4r{propertyr0r)r rrrqbytesr classmethodr rtyper!rcrrrdrirrprr(rrrrfrrrrrrrrrrrLrrrrr2r2r2r3rHs, H ! ) )( &$     $  rHqu) type_prefix)Arsrrrcollections.abcrr dataclassesrrtypingrrrgoogle.protobuf.messager grpclibr synchronicityr synchronicity.async_wrapr modal_protor _load_contextr_objectrrrrr _resolverr_serializationrr_utils.async_utilsrrr_utils.deprecationrr_utils.grpc_utilsr_utils.name_utilsr_utils.time_utilsrr r7r! exceptionr"r#r$r%r&r'r(r4 QueueManagerrHr9r2r2r2r3sF              #