grocery-manager/app/rest/CachedLoadingPage.py

#
# Copyright (c) Daniel Sheffield 2023
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
from queue import Queue, Empty
from time import time
from threading import Lock
from typing import Callable, Union

STALE = 10*60

class CachedLoadingPage():
    
    value: str

    def __init__(self, initial_value: Union[str, list], provider: Callable[[Queue], None], incremental: bool = True):
        self._created = time()
        self._queue = Queue()
        self._loaded = False
        self.value = initial_value
        self._lock = Lock()
        self.provider = provider
        self.incremental = incremental

    @property
    def age(self) -> float:
        return time() - self._created

    @property
    def queue(self) -> Queue:
        return self._queue

    @property
    def loaded(self) -> bool:
        return self._loaded

    def _set_loaded(self, value: bool) -> bool:
        self._loaded = value
        return self._loaded
    
    @property
    def stale(self) -> bool:
        return self.age > STALE
    
    def _start(self) -> None:
        if not self.provider:
            return

        self.provider(self.queue)
        self.provider = None

    def update(self) -> Union[str, list]:
        if not self._lock.acquire(blocking=True, timeout=0.5):
            return self.value
        try:
            self._start()
            item = self._queue.get(block=True, timeout=0.5)
            if item is None:
                self._queue.task_done()
                self._set_loaded(True)
            else:
                if self.incremental:
                    self.value.append(item)
                else:
                    self.value = item
                self.queue.task_done()
        except Empty:
            pass
        finally:
            self._lock.release()
        return self.value