Function get_background_callback_manager

tests/background_callback/utils.py:34–66

Get the long callback mangaer configured by environment variables

()

def get_background_callback_manager():
    """
    Get the long callback mangaer configured by environment variables
    """
    if os.environ.get("LONG_CALLBACK_MANAGER", None) == "celery":
        from dash.background_callback import CeleryManager
        from celery import Celery
        import redis

        celery_app = Celery(
            __name__,
            broker=os.environ.get("CELERY_BROKER"),
            backend=os.environ.get("CELERY_BACKEND"),
        )
        background_callback_manager = CeleryManager(celery_app)
        redis_conn = redis.Redis(host="localhost", port=6379, db=1)
        background_callback_manager.test_lock = redis_conn.lock("test-lock")
    elif os.environ.get("LONG_CALLBACK_MANAGER", None) == "diskcache":
        import diskcache

        cache = diskcache.Cache(os.environ.get("DISKCACHE_DIR"))
        background_callback_manager = TestDiskCacheManager(cache)
        background_callback_manager.test_lock = diskcache.Lock(cache, "test-lock")
    else:
        raise ValueError(
            "Invalid long callback manager specified as LONG_CALLBACK_MANAGER "
            "environment variable"
        )

    global manager
    manager = background_callback_manager

    return background_callback_manager


def kill(proc_pid):