or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

cli-parsing.mdcollections.mdconfiguration.mdindex.mdsubprocess-runners.mdtask-execution.mdwatchers.md
tile.json

tessl/pypi-invoke

Pythonic task execution library for managing shell-oriented subprocesses and organizing executable Python code into CLI-invokable tasks

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/invoke@2.2.x

To install, run

npx @tessl/cli install tessl/pypi-invoke@2.2.0

index.mddocs/

Invoke

Invoke is a Python library for managing shell-oriented subprocesses and organizing executable Python code into CLI-invokable tasks. It provides a powerful framework for creating command-line interfaces, executing shell commands with proper error handling and output capture, organizing tasks into collections, and managing configuration through multiple sources.

Package Information

  • Package Name: invoke
  • Language: Python
  • Installation: pip install invoke

Core Imports

import invoke

Common imports for basic usage:

from invoke import task, run, Context, Collection

Full import for comprehensive functionality:

from invoke import (
    # Core execution
    run, sudo, Context, MockContext,
    # Task management  
    task, call, Task, Call, Collection,
    # Configuration
    Config,
    # CLI and program management
    Program, Executor,
    # Subprocess execution
    Runner, Local, Result, Promise, Failure,
    # CLI parsing
    Parser, Argument, ParserContext, ParseResult,
    # Collection loading
    FilesystemLoader,
    # Stream watchers
    StreamWatcher, Responder, FailingResponder,
    # Utilities
    pty_size,
    # Exceptions
    Exit, ParseError, UnexpectedExit, CollectionNotFound, 
    CommandTimedOut, AuthFailure, WatcherError
)

Basic Usage

Simple Command Execution

from invoke import run

# Run a simple command
result = run("ls -la")
print("Return code:", result.return_code)
print("Output:", result.stdout)

# Run with sudo
result = run("sudo service nginx restart", pty=True)

Task Definition and Execution

from invoke import task, Collection

@task
def hello(ctx, name="World"):
    """Say hello to someone."""
    print(f"Hello {name}!")

@task  
def deploy(ctx, env="staging"):
    """Deploy application to environment."""
    ctx.run(f"git push {env} main")
    ctx.run(f"ssh {env} 'systemctl restart app'")

# Create collection and run
ns = Collection(hello, deploy)

Context-based Execution

from invoke import Context

# Create context for command execution
ctx = Context()

# Execute commands through context
result = ctx.run("echo 'Hello World'", hide=True)
print(result.stdout)

# Change working directory
with ctx.cd('/tmp'):
    ctx.run("pwd")  # Shows /tmp

# Use command prefixes
with ctx.prefix('source venv/bin/activate'):
    ctx.run("python --version")

Architecture

Invoke's architecture is built around several key concepts:

  • Tasks: Decorated Python functions that can be invoked from CLI
  • Context: Execution environment providing subprocess management and configuration
  • Collections: Hierarchical task organization and namespacing
  • Runners: Pluggable command execution strategies (Local, SSH, etc.)
  • Configuration: Multi-source config system (files, env vars, CLI args)
  • Watchers: Stream processors for interactive subprocess communication

Capabilities

Task Definition and Execution

Core functionality for defining Python functions as invokable tasks, with argument parsing, help generation, and execution management.

def task(*args, **kwargs): ...
def call(task, *args, **kwargs) -> Call: ...

class Task:
    def __init__(self, body, name=None, aliases=None, positional=None, optional=None, default=False, auto_shortflags=True, help=None, pre=None, post=None, iterable=None, incrementable=None): ...

class Call:
    def __init__(self, task, args=None, kwargs=None): ...
    def make_context(self, config): ...

Task Definition and Execution

Subprocess Command Execution

Command execution through contexts with support for various runners, output capture, error handling, and interactive features.

def run(command: str, **kwargs) -> Result: ...
def sudo(command: str, **kwargs) -> Result: ...

class Context:
    def run(self, command, **kwargs): ...
    def sudo(self, command, **kwargs): ...
    def cd(self, path): ...
    def prefix(self, command): ...

class Result:
    return_code: int
    stdout: str
    stderr: str
    ok: bool
    failed: bool

Subprocess Command Execution

Configuration Management

Multi-source configuration system supporting files, environment variables, and command-line arguments with hierarchical merging.

class Config:
    def __init__(self, defaults=None, overrides=None, lazy=False, runtime_path=None, system_prefix=None, user_prefix=None, project_location=None, env_prefix=None, file_prefix=None): ...
    def load_defaults(self, merge=True): ...
    def merge(self, *dicts): ...
    def clone(self, into=None): ...

Configuration Management

CLI Argument Parsing

Comprehensive argument parsing system supporting flags, options, positional arguments, and help generation.

class Parser:
    def __init__(self, contexts=None, initial=None, ignore_unknown=False): ...
    def parse_argv(self, argv): ...

class Argument:
    def __init__(self, names=None, kind=str, default=None, help=None, positional=False, optional=None, incrementable=False, iterable=False, attr_name=None): ...

class ParseResult:
    def __getitem__(self, index): ...
    def __iter__(self): ...

CLI Argument Parsing

Task Collections

Hierarchical organization of tasks into collections with namespace management, auto-discovery, and configuration integration.

class Collection:
    def __init__(self, *args, **kwargs): ...
    def add_task(self, task, name=None, aliases=None, default=None): ...
    def add_collection(self, coll, name=None): ...
    @classmethod
    def from_module(cls, module, name=None, config=None, loaded_from=None, auto_dash_names=True): ...

Task Collections

Stream Watchers

Interactive subprocess communication through pattern-based responders and stream processors.

class StreamWatcher:
    def submit(self, stream): ...

class Responder(StreamWatcher):
    def __init__(self, pattern, response, sentinel=None): ...
    def pattern_matches(self, stream, pattern): ...

class FailingResponder(Responder):
    def __init__(self, pattern, response, sentinel=None): ...

Stream Watchers

Types

Core Types

class Context:
    config: Config
    cwd: str

class MockContext(Context):
    pass

class Task:
    name: str
    called: bool
    times_called: int

class Call:
    task: Task
    args: tuple
    kwargs: dict

class Collection:
    task_names: list

class Config:
    pass

class Result:
    return_code: int
    stdout: str  
    stderr: str
    ok: bool
    failed: bool
    
class Promise(Result):
    pass

class Failure:
    result: Result
    reason: str

Exception Types

class Exit(SystemExit):
    code: int

class ParseError(ValueError):
    pass

class CollectionNotFound(ImportError):
    pass

class UnexpectedExit(Exception):
    result: Result

class CommandTimedOut(Exception):
    timeout: float

class AuthFailure(Exception):
    result: Result

class WatcherError(Exception):
    pass

class ResponseNotAccepted(Exception):
    pass

class SubprocessPipeError(Exception):
    pass

class ThreadException(Exception):
    pass

class PlatformError(OSError):
    pass

class AmbiguousEnvVar(Exception):
    pass

class UncastableEnvVar(Exception):
    pass

class UnknownFileType(Exception):
    pass

class UnpicklableConfigMember(Exception):
    pass