import modal import queue from dataclasses import dataclass from datetime import datetime from threading import Thread INBOUND_EVENT_QUEUE_NAME = "inbound_event_queue" from enum import Enum class EventType(Enum): CANCEL = "cancel" @dataclass class Event: item_id: str event_type: EventType time_sent = datetime.now() class InboundEventQueue(Thread): """ A thread that streams events from a modal queue to a local queue. """ def __init__(self, modal_queue: modal.Queue, item_id: str, timeout: int = 60 * 8): self.local_queue: queue.Queue[Event] = queue.Queue() self.modal_queue = modal_queue self.item_id = item_id self.timeout = timeout super().__init__() self.daemon = True def fill_local_queue(self): # Moves items from modal queue to local queue. for event in self.modal_queue.iterate(partition=self.item_id, item_poll_timeout=self.timeout): assert isinstance(event, Event) self.local_queue.put(event) def run(self): self.fill_local_queue()