Cache

Cache

Bases: Struct

Cache class for handling file-based caching with support for asynchronous data loading.

Attributes:

Name	Type	Description
cache_path	str	The path where the cache file is stored.
method	Optional[Callable]	A callable, typically an async method, to fetch data when the cache doesn't exist or needs to be reloaded.
date_as_suffix	bool	If True, appends today's date to the cache file name as a suffix.

Methods:

Name	Description
get_async	Asynchronously retrieves data from the cache. If the cache file doesn't exist, it will fetch data using the provided method and store it in the cache.
clean_cache	Deletes the cache file from disk.

Usage:

async def my_function():
    await asyncio.sleep(5)
    return []

cache = Cache(
    cache_path=f"./cache/markets_{ex.exchange_name}.json", method=ex.get_contracts
)
data = await cache.get_async()

# Use this if you don't have any other async.io task, otherwise the program will end before `my_function` runs
await cache.wait_till_complete()

cache_path

cache_path: str

Where you want the file to be stored

date_as_suffix

date_as_suffix: bool = False

If true, will append today's date to the cache file.

dump_format

dump_format: FileFormat = JSON

Format of the file to dump to. It might be None if your method already dumps the file.

file_format

file_format: FileFormat = JSON

Format of the file to read.

method

method: Optional[Callable] = None

A lambda method to run in case the file doesn't exist

clean_cache

clean_cache()

Removes the cache file.

Source code in src/tradingtoolbox/utils/cache.py

def clean_cache(self):
    """
    Removes the cache file.
    """
    if os.path.exists(self.cache_path):
        os.remove(self.cache_path)

get_async

get_async(reload=True, **method_kwargs)

Gets the data from the cache. If it doesn't exist, it will reload it.

Parameters:

Name	Type	Description	Default
reload	bool	Whether to reload the data. Defaults to True.	True
**method_kwargs		Keyword arguments to pass to the method.	{}

Returns:

Name	Type	Description
dict		The data from the cache.

Source code in src/tradingtoolbox/utils/cache.py

async def get_async(self, reload=True, **method_kwargs):
    """
    Gets the data from the cache. If it doesn't exist, it will reload it.

    Args:
        reload (bool, optional): Whether to reload the data. Defaults to True.
        **method_kwargs: Keyword arguments to pass to the method.

    Returns:
        dict: The data from the cache.
    """
    data = self._read_file()
    if data is not None:
        if reload:
            self._task = asyncio.create_task(self._reload_async(**method_kwargs))
    else:
        await self._reload_async(**method_kwargs)
        data = self._read_file()
    return data  # type: ignore

wait_till_all_tasks_complete

wait_till_all_tasks_complete()

Waits for all tasks to complete. Use this if you don't have any other async.io task to avoid the program ending before the method is ran

Source code in src/tradingtoolbox/utils/cache.py

@staticmethod
async def wait_till_all_tasks_complete():
    """
    Waits for all tasks to complete.
    Use this if you don't have any other async.io task to avoid the program ending before the method is ran
    """
    while True:
        pending_tasks = []
        for task in asyncio.all_tasks():
            if not task.done():
                pending_tasks.append(task)
        if len(pending_tasks) == 1:
            break
        await asyncio.sleep(1)

wait_till_complete

wait_till_complete()

Waits for the task to complete. Use this if you don't have any other async.io task to avoid the program ending before the method is ran

Source code in src/tradingtoolbox/utils/cache.py

async def wait_till_complete(self):
    """
    Waits for the task to complete.
    Use this if you don't have any other async.io task to avoid the program ending before the method is ran
    """
    if self._task:
        while True:
            if self._task.done():
                break
            await asyncio.sleep(1)  # Check every second