queue

django_spire.metric.domain.statistic.queue

logger = logging.getLogger(__name__) module-attribute

FLUSH_BATCH_SIZE = 100 module-attribute

tracking_queue = StatisticTrackingQueue() module-attribute

StatisticTrackingQueue

Source code in django_spire/metric/domain/statistic/queue.py

def __init__(self, *, maxsize: int | None = None, start_worker: bool = True) -> None:
    self._queue: queue.Queue[str] = queue.Queue(
        maxsize=maxsize or getattr(settings, 'DJANGO_SPIRE_METRIC_TRACKING_QUEUE_MAXSIZE', 1000)
    )
    self._lock = threading.Lock()
    self._worker: threading.Thread | None = None
    self._start_worker = start_worker

enqueue

Source code in django_spire/metric/domain/statistic/queue.py

def enqueue(self, reference: str) -> bool:
    try:
        self._queue.put_nowait(reference)
    except queue.Full:
        logger.warning('Dropped metric tracking reference %r: queue is full', reference)
        return False

    if self._start_worker:
        self._ensure_worker()
    return True

flush

Source code in django_spire/metric/domain/statistic/queue.py

def flush(self) -> None:
    batch = self._drain_batch()
    while batch:
        StatisticTrackingService.track_many(batch)
        batch = self._drain_batch()