tractor_beam.utils package

tractor_beam.utils.config module

class tractor_beam.utils.config.Config(conf: str | dict | None = None)

Bases: object

create(config: dict = None)

This Python function creates a project directory based on a provided configuration dictionary.

Parameters:

config – The config parameter in the create method is a dictionary that contains

configuration settings for a project. This method checks if the config parameter is provided, parses the configuration, creates a project directory based on the parsed settings, and writes the configuration to a config.json file in :type config: dict :return: either a success message indicating that the project has been successfully created and the configuration has been saved in a JSON file, or a fatal error message indicating that the project already exists or the provided config schema does not match the requirements.

destroy(confirm: str = None)

load_conf(conf)

parse_conf(conf_dict: Dict[str, Any]) → Schema

save()

This function saves a configuration object to a JSON file. :return: The save method returns either the saved configuration object (self.conf) and the path where it was saved (conf_path), or it returns None if there was an error or if there was no configuration to save.

unbox(overwrite: bool = False)

The function unbox creates a project directory based on configuration settings, with an option to overwrite existing directory.

Parameters:

overwrite – The overwrite parameter in the unbox method is a boolean flag that

determines whether to overwrite an existing directory if it already exists. If overwrite is set to True and the directory at proj_path already exists, the method will delete the existing directory and create a, defaults to False :type overwrite: bool (optional) :return: The unbox method returns the result of the self.save() method.

class tractor_beam.utils.config.ConfigEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)

to_serializable(obj)

class tractor_beam.utils.config.CustomJob(func: str | None = None, headers: dict | None = None, types: list | None = None)

Bases: object

func: str | None = None

headers: dict | None = None

types: list | None = None

class tractor_beam.utils.config.Job(url: str, types: List[str] | None = None, tasks: List[str] | None = None, beacon: str | None = None, delay: float | None = None, custom: tractor_beam.utils.config.CustomJob | None = None)

Bases: object

beacon: str | None = None

custom: CustomJob | None = None

delay: float | None = None

tasks: List[str] | None = None

types: List[str] | None = None

url: str

class tractor_beam.utils.config.Schema(role: str, settings: tractor_beam.utils.config.Settings)

Bases: object

role: str

settings: Settings

class tractor_beam.utils.config.Settings(name: str, proj_dir: str, jobs: List[Job])

Bases: object

tractor_beam.utils.globals module

tractor_beam.utils.globals.all_dir_size(directories: list = None)

tractor_beam.utils.globals.check(path)

tractor_beam.utils.globals.check_headers(data)

tractor_beam.utils.globals.dateme(receipt: dict = None)

tractor_beam.utils.globals.dir_size(directory: str = None)

tractor_beam.utils.globals.fetch_to_write(state, attachment, headers, attachment_path, file_name, block_size, o=False)

tractor_beam.utils.globals.files(content: str = None, url: str = None, types: list = None)

tractor_beam.utils.globals.readthis(path: str = None)

tractor_beam.utils.globals.writeme(content, path: str = None)

tractor_beam.utils.tools module