Module data

Classes

class Cache (caching_cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], int, DataPipe, ...], None] = None, *, loading_cb: Callable[[int], CacheObject] = None, revalidate: Optional[datetime.datetime] = None, timeout: Optional[datetime.timedelta] = None, max_requests: Optional[int] = None, store: bool = True, revalidate_cb: Callable[[datetime.datetime, datetime.datetime, Tuple[Any, ...], PipeOutput[Any], dict[str, Any]], RevalidateAction] = None)

Function decorator designed to make a Cache pipe out of a function. The decorated function is expected to be the caching method. When using this, it is recommended to pass a load callback method to load the cache object from the storage. The caching method should be able to store the cache and load it from the storage. The cache object should be a CacheObject instance. It is also possible to pass a revalidate callback method to define a dynamic revalidate action. The revalidate action is a method that returns a RevalidateAction enum value. The revalidate action is used to determine if the cache should be revalidated or not. If no revalidate method is passed, the default revalidate method check if the cache is expired given an expired datetime, a timeout or a maximum number of requests.

Note

If timeout is None, max_requests is None and revalidate_cb is None, the cache will never be revalidated. If the structure of the underlying pipe changes or the parameters of the underlying pipe changes, the whole pipeline will be revalidated (full revalidation). This means that every cache pipe will revalidate their cache, not only the one that changed.

Examples:

>>> @Cache(loading_cb=JSON_load, store=True, timeout=timedelta(seconds=1))
... def MyCache(frm: datetime, to: datetime, *args, po: PipeOutput, pipe_id: int, revalidate: datetime,
...          timedelta, max_requests: int, n_requests: int, **kwargs):
...     value = {
...         "data": po.value,
...         "stored_dt": datetime.now().isoformat(),
...         "revalidate": revalidate.isoformat(),
...         "current_n_requests": n_requests,
...         "timeout": timeout.total_seconds(),
...         "max_requests": max_requests
...     }
...     if not os.path.exists(".cache"):
...         os.mkdir(".cache")
...     with open(f".cache/{pipe_id}.json", "w") as f:
...         json.dump(value, f)
>>> # The cache pipe can also be used as is:
>>> pipe = FetchN | Cache(timeout=timedelta(seconds=1))

Args:

caching_cb: The callback that is called after the pipeline wrapped by the Cache pipe is executed. It is used to store the cache.
loading_cb: The callback used to load the cache from the disk.
revalidate: The next datetime to revalidate the cache.
timeout: A timedelta object representing the max age of the cache before revalidating it. If it is oneday, the cache will be revalidated every day as opposed to the revalidate parameter, which revalidate only once.
max_requests: The maximum number of request that hit the cache before revalidating it.
store: Whether to store the cache on disk or keep it only in memory.
revalidate_cb: A callback method that returns a RevalidateAction enum value. It is used to determine ifthe cache should be revalidated or not. This callback can be provided for complex revalidation logic.

Parent

• DataPipe

class Collate (cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], PipeOutput, ...], Any])

Function decorator designed to make a Collate pipe out of a function. It is designed to collate data. The data can be combined, aggregated, concatenated, etc. The pipe is named after the function name.

Args:

cb: The collate callback method.

Parent

• DataPipe

class Fetch (cb: Callable[[datetime.datetime, datetime.datetime, ..., Optional[PipeOutput[Any]], ...], Any])

Function decorator designed to make a Fetch pipe out of a function. It is designed to fetch data from a source. The source can be a database, a web API, a file, etc. The pipe is named after the function name.

Args:

cb: The fetch callback method.

Parent

• DataPipe

class Process (cb: Callable[[datetime.datetime, datetime.datetime, ..., PipeOutput[Any], ...], Any])

Function decorator designed to make a Process pipe out of a function. It is designed to process data. The data can be transformed, cleaned, imputed, etc. The pipe is named after the function name.

Args:

cb: The process callback method.

Parent

• DataPipe