ShellProxy

Proxy for executing shell commands with advanced output handling and logging.

This class provides a high-level interface for running shell commands in a subprocess with features including:

Real-time output dispatching to message observers
Environment variable management
Working directory control
Both blocking and non-blocking execution modes
Integration with progress bars and message dispatchers

The proxy can be used as a context manager for automatic cleanup of temporary directories.

Logging Format Support

When using ShellProxy with a message_dispatcher (e.g., in virtual environment agents), the stdout/stderr output supports special format prefixes for message categorization:

[INFO] message - Info message
[SUCCESS] message - Success message
[WARNING] message - Warning message
[ERROR] message - Error message
[DEBUG] message - Debug message
[PROGRESS:XX] message - Progress message (XX = 0-100)

These format prefixes are automatically parsed by the message_dispatcher to dispatch messages with the appropriate severity level and type. This is particularly useful in virtual environment agents where standard logging methods are not available.

Example: python # In a virtual environment agent script print("[INFO] Starting processing...") print("[PROGRESS:50] Halfway done...") print("[SUCCESS] Processing completed!")

Attributes

working_dir: str

Functions

__init__

Initialize a ShellProxy instance.

working_dir : str | None

Working directory for the shell (all commands will be executed from this dir). If not provided, a new temporary directory is created

message_dispatcher : gws_core.core.classes.observer.message_dispatcher.MessageDispatcher | None

If provided, the output of the command will be redirected to the dispatcher. In a task it can be provided like this: ShellProxy(message_dispatcher=self.message_dispatcher) Can be useful to log command outputs in task's logs

attach_observer

Attach a custom observer to the shell proxy.

The logs of the proxy will be dispatched to the observer.

observer : MessageObserver

Observer to attach

attach_progress_bar

Attach a progress bar to the shell proxy.

The logs of the proxy will be dispatched to the progress bar logs.

progress_bar : ProgressBar

Progress bar to attach

check_output

Run a command in a shell and return the output. The logs of the command are ignored.

cmd : list | str

command to run

env : dict | None

environment variables to pass to the shell

env_mode : ShellProxyEnvVariableMode - ShellProxyEnvVariableMode.MERGE

mode to use for the environment variables

shell_mode : bool | None

if True, the command is run in a shell. If False, run without shell. If None (default), automatically detect from cmd type (string -> True, list -> False)

text : bool - True

if True, the output is returned as a string

Return type : Any

clean_working_dir

Delete the working directory and all its contents.

dispatch_waiting_messages

Force the message dispatcher to send all buffered messages immediately.

get_default_env_variables

Get the default environment variables for the shell proxy. To override in subclasses to provide custom additional default environment variables.

Return type : dict

get_message_dispatcher

Get the message dispatcher instance.

Return type : MessageDispatcher

log_error_message

Log an error message using the dispatcher.

message : str

Error message to log

log_info_message

Log an info message using the dispatcher.

message : str

Message to log

log_warning_message

Log a warning message using the dispatcher.

message : str

Warning message to log

run

Run a command in a shell. The logs of the command will be dispatched to the message dispatcher during the execution.

cmd : list | str

command to run

env : dict | None

environment variables to pass to the shell

env_mode : ShellProxyEnvVariableMode - ShellProxyEnvVariableMode.MERGE

mode to use for the environment variables

shell_mode : bool | None

if True, the command is run in a shell. If False, run without shell. If None (default), automatically detect from cmd type (string -> True, list -> False)

dispatch_stdout : bool - False

if True, the stdout of the command is dispatched to the message dispatcher. ⚠️ Warning ⚠️ Do not set to True if the command generates a lot of logs, because logs are stored in database

dispatch_stderr : bool - True

if True, the stderr of the command is dispatched to the message dispatcher. ⚠️ Warning ⚠️ Do not set to True if the command generates a lot of logs, because logs are stored in database

Return type : int

run_in_new_thread

Run a command in a shell without blocking the thread. The logs of the command are ignored.

cmd : list | str

command to run

env : dict | None

environment variables to pass to the shell

env_mode : ShellProxyEnvVariableMode - ShellProxyEnvVariableMode.MERGE

mode to use for the environment variables

shell_mode : bool | None

if True, the command is run in a shell. If False, run without shell. If None (default), automatically detect from cmd type (string -> True, list -> False)

Return type : SysProc

to_dto

Convert the ShellProxy instance to a data transfer object.

Return type : ShellProxyDTO

classname @classmethod

Returns the name of the class

slugify : bool - False

True to slugify the class name if True

snakefy : bool - False

True to snakefy the class name if True

replace_uppercase : bool - False

Replace upper cases by "-" if True

Return type : str

full_classname @classmethod

Returns the full name of the class

slugify : bool - False

Slugify the returned class name if True

snakefy : bool - False

Snakefy the returned class name if True

Return type : str

get_brick_name @classmethod

Return type : str

get_brick_version @classmethod

Return type : Version

get_human_name @classmethod

Get the human name of the object

Return type : str

get_short_description @classmethod

Get the short description of the object

Return type : str

get_style @classmethod

Get the style of the object

Return type : TypingStyle

get_typing_name @classmethod

Get the typing name of the object Ex: 'RESOURCE.gws_core.Table'

Return type : str

get_typing_name_obj @classmethod

Return type : TypingNameObj

inheritors @classmethod

Get all the classes that inherit this class

Return type : list

Logging Format Support

Have you developed a brick?