helpers
OptionalDependencyError
WorkflowSorter
Takes a collection of nodes + edges and returns a list of nodes ordered so that every node appears after all of its upstream dependencies.
Methods
__init__(nodes: List[WorkflowNode], edges: List[WorkflowEdge])
sorted_nodes() -> List[WorkflowNode]
Functions
package_version(package_name, include_name = True)
compress_files(path: str = '.', files: List = [], compression: int = zipfile.ZIP_DEFLATED, zip_filename: str = 'my.zip')
compress_folder(path: str = '.', compression: int = zipfile.ZIP_DEFLATED, zip_filename: str = 'my.zip')
save_pkl(data_folder, filename, contents, mode = 'wb', extension = '.pkl')
load_pkl(data_folder, filename, mode = 'rb', extension = '.pkl')
save_dill(folder, filename, contents, mode = 'wb', extension = '.pkl')
load_dill(folder, filename, mode = 'rb', extension = '.pkl')
save_anything(data_folder, filename, contents, mode = 'wb', protocol: Union['pkl', 'dill'] = 'pkl', extension = '.pkl')
load_anything(data_folder, filename, mode = 'rb', protocol: Union['pkl', 'dill'] = 'pkl', extension = '.pkl')
save_json(data_folder, filename, contents, mode = 'w', options = {'indent': 4}, extension = '.json', force_folder = True)
Saves a dictionary as a JSON file.
:param data: Dictionary to be saved. :param filename: Name of the output file.
load_json(data_folder, filename, mode = 'r', options = {}, extension = '.json')
Loads a JSON file into a dictionary.
:param filename: Name of the input file. :return: Dictionary loaded from the JSON file.
save_jsonl(data_folder, filename, data_list: List = [], extension: str = '.jsonl')
Saves a list of dictionaries as a JSONL file.
:param data_list: List of dictionaries to be saved. :param filename: Name of the output file.
load_jsonl(data_folder, filename, extension = '.jsonl')
Loads a JSONL file into a list of dictionaries.
:param filename: Name of the input file. :return: List of dictionaries loaded from the JSONL file.