Właśnie wypróbowałem Pythonaodłożyć na półkę moduł jako trwała pamięć podręczna dla danych pobranych z usługi zewnętrznej.Kompletny przykład jest tutaj.

Zastanawiałem się, jakie byłoby najlepsze podejście, jeśli chcę uczynić ten proces wieloprocesowy bezpiecznym? Wiem o redis, memcached i takich „prawdziwych rozwiązaniach”, ale chciałbym używać tylko części standardowej biblioteki Pythona lub bardzo minimalnych zależności, aby mój kod był zwarty i nie wprowadzał niepotrzebnej złożoności podczas uruchamiania kodu w pojedynczym procesie - model z pojedynczą nicią.

Łatwo jest wymyślić rozwiązanie jednoprocesowe, ale to nie działa dobrze w obecnych czasach uruchamiania Pythona. W szczególności problemem może być środowisko Apache + mod_wsgi

Tylko jeden proces jednorazowo aktualizuje buforowane dane (jakoś blokuje plik?)

Inne procesy wykorzystują buforowane dane w trakcie aktualizacji

Jeśli w procesie nie uda się zaktualizować buforowanych danych, naliczana jest kara w wysokości N minut, zanim inny proces będzie mógł spróbować ponownie (aby zapobiecgrzmiące stado i takie) - jak zasygnalizować to między procesami mod_wsgi

Nie używasz do tego żadnych „ciężkich narzędzi”, tylko standardowe biblioteki Pythona i UNIX

Również jeśli jakiś pakiet PyPi robi to bez zewnętrznych zależności, daj mi znać. Mile widziane są alternatywne podejścia i zalecenia, takie jak „po prostu użyj sqlite”.

Przykład:

import datetime
import os
import shelve
import logging


logger = logging.getLogger(__name__)


class Converter:

    def __init__(self, fpath):
        self.last_updated = None
        self.data = None

        self.data = shelve.open(fpath)

        if os.path.exists(fpath):
            self.last_updated = datetime.datetime.fromtimestamp(os.path.getmtime(fpath))

    def convert(self, source, target, amount, update=True, determiner="24h_avg"):
        # Do something with cached data
        pass

    def is_up_to_date(self):
        if not self.last_updated:
            return False

        return datetime.datetime.now() < self.last_updated + self.refresh_delay

    def update(self):
        try:
            # Update data from the external server
            self.last_updated = datetime.datetime.now()
            self.data.sync()
        except Exception as e:
            logger.error("Could not refresh market data: %s %s", self.api_url, e)
            logger.exception(e)