Write after timing or accumulation.

writecache.py

import abc
import threading
from typing import Any


class WriteCache(metaclass=abc.ABCMeta):
    def __init__(self, max_size: int = 100, min_time: float = 1) -> None:
        self.max_size = max_size
        self.min_time = min_time

        self.__size = max_size
        self.__lock = threading.RLock()
        self.__event = threading.Event()

        # Start timing thread
        self.set_writed()

    def timing_handle(self) -> None:
        if self.__event.wait(self.min_time):
            return
        with self.__lock:
            try:
                self.real_write()
            except:
                pass
            finally:
                self.set_writed()

    def write(self, data: Any) -> None:
        with self.__lock:
            self.__size -= 1
            try:
                return self._write(data)
            finally:
                if self.should_write:
                    try:
                        self.real_write()
                    finally:
                        self.set_writed()

    @property
    def should_write(self) -> bool:
        return self.__size <= 0

    def set_writed(self) -> None:
        self.__size = self.max_size
        self.__event.set()
        self.__event.clear()
        threading.Thread(target=self.timing_handle, daemon=True).start()

    @abc.abstractmethod
    def _write(self, data: Any) -> None:
        raise NotImplementedError

    @abc.abstractmethod
    def real_write(self) -> None:
        raise NotImplementedError