backend/tile_server/app/services/slide_cache.py

from collections import OrderedDict
from threading import Lock
from typing import Dict, Optional

import openslide


class SlideCache:
    def __init__(self, max_items: int = 8) -> None:
        self._max_items = max_items
        self._slides: Dict[str, openslide.OpenSlide] = OrderedDict()
        self._lock = Lock()

    def get(self, slide_id: str) -> Optional[openslide.OpenSlide]:
        with self._lock:
            slide = self._slides.get(slide_id)
            if slide is None:
                return None
            self._slides.move_to_end(slide_id)
            return slide

    def put(self, slide_id: str, path: str) -> openslide.OpenSlide:
        with self._lock:
            if slide_id in self._slides:
                self._slides.move_to_end(slide_id)
                return self._slides[slide_id]

            slide = openslide.OpenSlide(path)
            self._slides[slide_id] = slide
            self._slides.move_to_end(slide_id)

            while len(self._slides) > self._max_items:
                _, oldest = self._slides.popitem(last=False)
                try:
                    oldest.close()
                except Exception:
                    pass

            return slide


cache = SlideCache()


def load_slide(slide_id: str, path: str) -> openslide.OpenSlide:
    return cache.put(slide_id, path)


def get_slide(slide_id: str) -> Optional[openslide.OpenSlide]:
    return cache.get(slide_id)