Collection

Extend Invoke for Calcipy.

Classes

Collection

Bases: Collection

Calcipy Task Collection.

Source code in calcipy/collection.py

class Collection(InvokeCollection):
    """Calcipy Task Collection."""

    @classmethod
    def from_module(
        cls,
        module: ModuleType,
        name: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        loaded_from: Optional[str] = None,
        auto_dash_names: Optional[bool] = None,
    ) -> InvokeCollection:
        """Extend search for a namespace, Task, or deferred task.

        Returns:
            Collection populated with tasks from the module.

        """
        collection = super().from_module(
            module=module,
            name=name,
            config=config,
            loaded_from=loaded_from,
            auto_dash_names=auto_dash_names,
        )

        # If tasks were not loaded from a namespace or otherwise found
        if not collection.task_names:
            # Look for any decorated, but deferred "Tasks"
            for task in (fxn for fxn in vars(module).values() if hasattr(fxn, TASK_ARGS_ATTR)):
                collection.add_task(task)

        return collection

    def add_task(
        self,
        task: DeferredTask,
        name: Optional[str] = None,
        aliases: Optional[Tuple[str, ...]] = None,
        default: Optional[bool] = None,
    ) -> None:
        """Extend for deferred tasks."""
        super().add_task(task=_build_task(task), name=name, aliases=aliases, default=default)

Functions

add_task

add_task(task, name=None, aliases=None, default=None)

Extend for deferred tasks.

Source code in calcipy/collection.py

def add_task(
    self,
    task: DeferredTask,
    name: Optional[str] = None,
    aliases: Optional[Tuple[str, ...]] = None,
    default: Optional[bool] = None,
) -> None:
    """Extend for deferred tasks."""
    super().add_task(task=_build_task(task), name=name, aliases=aliases, default=default)

from_module classmethod

from_module(module, name=None, config=None, loaded_from=None, auto_dash_names=None)

Extend search for a namespace, Task, or deferred task.

RETURNS	DESCRIPTION
Collection	Collection populated with tasks from the module.

Source code in calcipy/collection.py

@classmethod
def from_module(
    cls,
    module: ModuleType,
    name: Optional[str] = None,
    config: Optional[Dict[str, Any]] = None,
    loaded_from: Optional[str] = None,
    auto_dash_names: Optional[bool] = None,
) -> InvokeCollection:
    """Extend search for a namespace, Task, or deferred task.

    Returns:
        Collection populated with tasks from the module.

    """
    collection = super().from_module(
        module=module,
        name=name,
        config=config,
        loaded_from=loaded_from,
        auto_dash_names=auto_dash_names,
    )

    # If tasks were not loaded from a namespace or otherwise found
    if not collection.task_names:
        # Look for any decorated, but deferred "Tasks"
        for task in (fxn for fxn in vars(module).values() if hasattr(fxn, TASK_ARGS_ATTR)):
            collection.add_task(task)

    return collection

GlobalTaskOptions dataclass

Global Task Options.

Source code in calcipy/collection.py

@dataclass
class GlobalTaskOptions:
    """Global Task Options."""

    working_dir: Path = field(default_factory=Path.cwd)
    """Working directory for the program to use globally."""

    file_args: List[Path] = field(default_factory=list)
    """List of Paths to modify."""

    verbose: int = field(default=0)
    """Verbosity level."""

    keep_going: bool = False
    """Continue task execution regardless of failure."""

    capture_output: bool = False
    """Capture stdout and stderr output from tasks."""

    def __post_init__(self) -> None:
        """Validate dataclass."""
        options_verbose = [*LOG_LOOKUP.keys()]
        if self.verbose not in options_verbose:
            error = f'verbose must be one of: {options_verbose}'
            raise ValueError(error)

Attributes

capture_output class-attribute instance-attribute

capture_output = False

Capture stdout and stderr output from tasks.

file_args class-attribute instance-attribute

file_args = field(default_factory=list)

List of Paths to modify.

keep_going class-attribute instance-attribute

keep_going = False

Continue task execution regardless of failure.

verbose class-attribute instance-attribute

verbose = field(default=0)

Verbosity level.

working_dir class-attribute instance-attribute

working_dir = field(default_factory=cwd)

Working directory for the program to use globally.

Functions

__post_init__

__post_init__()

Validate dataclass.

Source code in calcipy/collection.py

def __post_init__(self) -> None:
    """Validate dataclass."""
    options_verbose = [*LOG_LOOKUP.keys()]
    if self.verbose not in options_verbose:
        error = f'verbose must be one of: {options_verbose}'
        raise ValueError(error)