Shell Command API

This section documents the shell command integration API.

ShellCommand Class

class flow.shell.ShellCommand(command, *, input_mode='args', output_mode='lines', cwd=None, env=None, shell=False, timeout=None, check=True, encoding='utf-8', errors='strict')[source]

Bases: object

A shell command that can be used as a node in a flow.

This class provides flexible ways to integrate shell commands: - As a source: Command output lines are streamed as items - As a transform: Input is passed via stdin, output via stdout - As a sink: Input is passed as arguments or stdin

Examples

# As a source - stream log file lines >>> cmd = ShellCommand(“tail -f /var/log/app.log”) >>> await flow().source(cmd, str).transform(parse_log, LogEntry).sink(store_log).execute()

# As a transform - use grep to filter >>> grep = ShellCommand(“grep ERROR”, input_mode=”stdin”) >>> await flow().source(logs, str).transform(grep, str).sink(print).execute()

# As a sink - append to file >>> append = ShellCommand(“tee -a output.log”, input_mode=”stdin”) >>> await flow().source(data, str).sink(append).execute()

Parameters:
__init__(command, *, input_mode='args', output_mode='lines', cwd=None, env=None, shell=False, timeout=None, check=True, encoding='utf-8', errors='strict')[source]

Initialize a shell command.

Parameters:

  • command (Union[str, List[str]]) – Command to run (string or list of arguments)

  • input_mode (str, default: 'args') – How to pass input (“args”, “stdin”, or “none”)

  • output_mode (str, default: 'lines') – How to capture output (“lines”, “all”, or “none”)

  • cwd (Union[str, Path, None], default: None) – Working directory for the command

  • env (Optional[Dict[str, str]], default: None) – Environment variables (None means inherit)

  • shell (bool, default: False) – Whether to run through shell (security risk!)

  • timeout (Optional[float], default: None) – Command timeout in seconds

  • check (bool, default: True) – Whether to raise on non-zero exit

  • encoding (str, default: 'utf-8') – Text encoding for stdin/stdout

  • errors (str, default: 'strict') – How to handle encoding errors

async run(input_data=None)[source]

Run the command once with optional input.

Parameters:

input_data (Optional[str], default: None)

Return type:

ShellResult

async stream_output()[source]

Stream command output line by line (for use as source).

Return type:

AsyncIterator[str]

async __call__(input_item)[source]

Make the command callable for use as a transform or sink.

Parameters:

input_item (TypeVar(T))

Return type:

Union[str, ShellResult, None]

ShellResult Class

class flow.shell.ShellResult(stdout, stderr, returncode, command)[source]

Bases: object

Result of a shell command execution.

Parameters:
__init__(stdout, stderr, returncode, command)[source]
Parameters:

Factory Functions

shell_source

flow.shell.shell_source(command, output_type=<class 'str'>, *, cwd=None, env=None, shell=False, encoding='utf-8', parser=None)[source]

Create a source node from a shell command.

The command’s stdout is streamed line by line as flow items.

Parameters:

  • command (Union[str, List[str]]) – Command to run

  • output_type (type[TypeVar(T)], default: <class 'str'>) – Type of output items (default: str)

  • cwd (Union[str, Path, None], default: None) – Working directory

  • env (Optional[Dict[str, str]], default: None) – Environment variables

  • shell (bool, default: False) – Whether to use shell

  • encoding (str, default: 'utf-8') – Output encoding

  • parser (Optional[Callable[[str], TypeVar(T)]], default: None) – Optional function to parse each line

Return type:

Callable[[], AsyncIterator[TypeVar(T)]]

Returns:

An async generator function suitable for use with .source()

Examples

# Stream system logs >>> logs = shell_source(“tail -f /var/log/syslog”) >>> await flow().source(logs, str).sink(print).execute()

# Parse structured data >>> def parse_json(line: str) -> dict: … return json.loads(line) >>> events = shell_source(“my-event-stream”, parser=parse_json) >>> await flow().source(events, dict).sink(process_event).execute()

shell_transform

flow.shell.shell_transform(command, output_type=<class 'str'>, *, input_mode='stdin', cwd=None, env=None, shell=False, timeout=None, encoding='utf-8', parser=None)[source]

Create a transform node from a shell command.

Input items are passed to the command via stdin or args, and stdout is returned as the transformed output.

Parameters:

  • command (Union[str, List[str]]) – Command to run

  • output_type (type[TypeVar(U)], default: <class 'str'>) – Type of output items

  • input_mode (str, default: 'stdin') – How to pass input (“stdin” or “args”)

  • cwd (Union[str, Path, None], default: None) – Working directory

  • env (Optional[Dict[str, str]], default: None) – Environment variables

  • shell (bool, default: False) – Whether to use shell

  • timeout (Optional[float], default: None) – Command timeout

  • encoding (str, default: 'utf-8') – I/O encoding

  • parser (Optional[Callable[[str], TypeVar(U)]], default: None) – Optional function to parse output

Return type:

Callable[[TypeVar(T)], TypeVar(U)]

Returns:

An async function suitable for use with .transform()

Examples

# Use jq to transform JSON >>> jq_filter = shell_transform(“jq ‘.data’”, dict, parser=json.loads) >>> await flow().source(json_docs, str).transform(jq_filter, dict).sink(save).execute()

# Use sed for text transformation >>> clean = shell_transform(“sed ‘s/[^a-zA-Z0-9 ]//g’”) >>> await flow().source(texts, str).transform(clean, str).sink(print).execute()

shell_sink

flow.shell.shell_sink(command, *, input_mode='stdin', cwd=None, env=None, shell=False, timeout=None, encoding='utf-8', check=True)[source]

Create a sink node from a shell command.

Input items are passed to the command, and the command is executed for its side effects (no output is returned).

Parameters:

  • command (Union[str, List[str]]) – Command to run

  • input_mode (str, default: 'stdin') – How to pass input (“stdin”, “args”, or “none”)

  • cwd (Union[str, Path, None], default: None) – Working directory

  • env (Optional[Dict[str, str]], default: None) – Environment variables

  • shell (bool, default: False) – Whether to use shell

  • timeout (Optional[float], default: None) – Command timeout

  • encoding (str, default: 'utf-8') – Input encoding

  • check (bool, default: True) – Whether to raise on non-zero exit

Return type:

Callable[[TypeVar(T)], None]

Returns:

An async function suitable for use with .sink()

Examples

# Append to file >>> append_log = shell_sink(“tee -a app.log”) >>> await flow().source(events, str).sink(append_log).execute()

# Send notifications >>> notify = shell_sink(“notify-send”, input_mode=”args”) >>> await flow().source(alerts, str).sink(notify).execute()