Module data

Functions

def toHash(obj) ‑> str

Hash an object. The hash is consistent across runs and is deterministic given the object.

Note

This function doesn't support all types of objects. This function is recursive and will hash the object recursively.

Args:

obj: The object to has

Returns:

The object hash

Classes

class CacheObject (value: Any, pipe_id: int, pipe_hash: str, next_revalidate: Optional[datetime.datetime] = None, max_request: Optional[int] = None, current_n_requests: int = 0)

This class holds a the output of a pipe and some metadata to be able to revalidate the cache.

Args:

value: The output of the pipe
pipe_id: The pipe_id of the pipe that generated the cache
pipe_hash: Used to check if the pipe structure has changed
next_revalidate: The next time the cache should be revalidated
max_request: The maximum number of requests that can be made before revalidating the cache
current_n_requests: The current number of requests made to the cache

Static methods

def load(pipe_id: int) ‑> CacheObject

Call this method to load a cache from the .cache folder.

Args:

pipe_id: The pipe_id that stored the cache

Returns:

The cache object if it exists else None. (This function doesn't check if the cache valid)

Methods

def store(self)

Call this method to store the cache to the .cache folder. The cache file has the pipe_id as its name. This is why it is strongly recommended to delete the .cache folder when changing the pipeline structure, to avoid unexpected behavior.

class DataPipe (T: DataPipeType, name: Optional[str] = None)

The base element of the data module api. This object can represent the four types of pipes (Fetch, Process, Cache, and Collate). It is a recursive object that can be used to build complex pipelines. The pipeline is built using the pipe operator (|). Once the pipeline is built, it is run by calling the get method. To make custom pipes, you need to inherit from this class and implement one of the three methods (fetch, process, collate). If you want to implement a cache pipe, you need to implement the load, cache and revalidate methods. Do not forget to call the super().init method in the init method of your custom pipe and pass the appropriate DataPipeType. You can also pass a name to the pipe to make the pipeline more readable.

Args:

T: The pipe type (Fetch, Process, Cache, Collate)
name: The pipe name. If None, the name of the class is used

Parent

  • abc.ABC

Static methods

def Collate(pipe1: DataPipe, pipe2: DataPipe) ‑> DataPipe

This method is used to build a collate pipe from two pipes. It is used top merge two branches of a pipeline.

Args:

pipe1: The first pipeline (or pipe if there is only one)
pipe2: The second pipeline (or pipe if there is only one)

Returns:

A new pipe that is the result of the concatenation of the two pipes. (Multiple pipes makes a pipeline)

Instance variables

var pipe_id : int

Get the pipe_id (Identifier unique to each pipe in the pipeline). It is given at built time and is deterministic. This means that each time the pipeline is built, the pipe_id will be the same for each pipe, given that the pipeline structure is the same.

Returns:

The pipe_id

Methods

def cache(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Any], **kwargs) ‑> None

This method is called to cache the output of the pipeline. It is called by the _run method when the pipe is a cache pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

None
def collate(self, frm: datetime.datetime, to: datetime.datetime, *args, po1: PipeOutput[typing.Any], po2: PipeOutput[typing.Any], **kwargs) ‑> PipeOutput

This method is called to merge the output of two pipes. It is called by the _run method when the pipe is a collate pipe.

Args:

frm: From datetime
to: To datetime
args: any args passed to the get method
po1: The output of the first pipeline (left)
po2: The output of the second pipeline (right)
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata
def fetch(self, frm: datetime.datetime, to: datetime.datetime, *args, po: Optional[PipeOutput[typing.Any]], **kwargs) ‑> PipeOutput

This method is called to fetch the data from an external source. It is called by the _run method when the pipe is a fetch pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata
def get(self, frm: datetime.datetime, to: datetime.datetime, *args, **kwargs) ‑> Any

Run the pipeline from the beginning to the end and return the output of the pipeline.

Args:

frm: From datetime, this is passed to all pipes and is used to fetch the data from this date to the 'to'date.
to: The end datetime, this is passed to all pipes and is used to fetch the data from the 'frm' date tothis date.
args: any args that needs to be passed to the subsequent pipes
kwargs: any keyword args that needs to be passed to the subsequent pipes

Returns:

The output of the pipeline
def hash(self) ‑> str

Make a hash that is consistent across runs and is deterministic given the object.

Returns:

The hash.
def load(self) ‑> Optional[CacheObject]

This method is called to load the cache from the disk. It is called by the _load_cache method when the pipe is a cache pipe. It will return the loaded cache object if found. Otherwise, it will return None.

Returns:

CacheObject if found else None
def process(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput[typing.Any], **kwargs) ‑> PipeOutput

This method is called to process and transform the data. It is called by the _run method when the pipe is a process pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata
def render(self)

This method is used to render the pipeline as a string. It renders the pipe structure as a tree with each node being the name of the pipe. It is used by the str method to render the pipeline as a tree.

Returns:

The rendered pipeline as a string
def revalidate(self, frm: datetime.datetime, to: datetime.datetime, *args, po: PipeOutput, **kwargs) ‑> RevalidateAction

This method is called to revalidate the cache. It is called by the _run method when the pipe is a cache pipe.

Args:

frm: From datetime
to: To datetime
args: Any args passed to the get method
po: The pipe output of the previous pipe
kwargs: Any keyword args passed to the get method

Returns:

A pipe output object that holds the output of the pipe and some metadata
def set_id(self, pipe_id: int)

With this method, it is possible to manually set the pipe_id of the pipe. It is recommended to call this method only on the top-level pipes, and before running the pipeline with get. If the pipe is not the top-level pipe, its pipe_id may change, and you might run in unexpected behaviors. By running this method only before running the get method, this ensures that the pipe will have the same pipe_id if the pipeline structure is the same. This is due to the fact that pipe_ids are dynamically assigned at build time. However, when the pipe is run, the pipe becomes forged and no changes are allowed. This ensures that the pipe ids stays the same. When this method is called, it acts as if the pipe was run, and the pipe_ids are frozen. This means that the ids aren't verified. Make sure that the pipe_id given are unique to avoid conflicts. Note: The passed pipe_id will be assigned to the lowes-level pipe. The current pipe will have the highest pipe_id

Args:

pipe_id: The pipe_id to assign to the pipe.

Returns:

None
class DataPipeType (*args, **kwds)

[Enum] - An enumeration representing the four types of pipes that can be used in the pipeline.

  • FETCH: Fetch data from an external source (Web, DB, file, API, etc.).
  • PROCESS: Process and transform the data (Filter, transform, etc.).
  • CACHE: Cache the output of the pipeline to avoid repeating previous steps on each run.
  • COLLATE: Collate the output of two pipes (This is a special case, used to merge two pipelines).

Parent

  • enum.Enum

Class variables

var CACHE
var COLLATE
var FETCH
var PROCESS
class PipeOutput (value: ~T_, output_from: DataPipe, revalidate_action: Optional[ForwardRef('RevalidateAction')] = None, **kwargs)

This class holds the output of a pipe and some metadata that are passed through out the pipeline. This metadata is used for revalidation purpose.

Args

value
The value
output_from
The output from the pipe
revalidate_action
The revalidate action

**kwargs:

Parent

  • typing.Generic

Instance variables

var revalidate

Get the revalidate action of the pipe

Returns:

The revalidate action of the pipe
var value

The output of the pipe

Returns:

The output value of the pipe

Methods

def set_output_from(self, output_from: DataPipe)

Set the output_from of the pipe. Can be useful when a pip makes a copy of the output of another pipe.

Args:

output_from: The pipe that generated the output

Returns:

None
def set_revalidate(self, action: RevalidateAction)

Set the revalidate action of the pipe.

Args:

action: The action to set

Returns:

None
class RevalidateAction (*args, **kwds)

The three possible revalidation action that a pipe can return.

  • REVALIDATE: Revalidate the cache from the current position to the end of the pipeline
  • FULL_REVALIDATE: Revalidate the cache from the beginning to the end of the pipeline
  • NOOP: Do not revalidate the cache (The pipe will load from cache if it exists)

Parent

  • enum.Enum

Class variables

var FULL_REVALIDATE
var NOOP
var REVALIDATE