commands

prefect_shell.commands

Tasks for interacting with shell commands

Functions

shell_run_command

shell_run_command(command: str, env: dict[str, str] | None = None, helper_command: str | None = None, shell: str | None = None, extension: str | None = None, return_all: bool = False, stream_level: int = logging.INFO, cwd: str | bytes | os.PathLike[str] | None = None) -> list[str] | str

Runs arbitrary shell commands.

Args:

• command: Shell command to be executed; can also be provided post-initialization by calling this task instance.
• env: Dictionary of environment variables to use for the subprocess; can also be provided at runtime.
• helper_command: String representing a shell command, which will be executed prior to the command in the same process. Can be used to change directories, define helper functions, etc. for different commands in a flow.
• shell: Shell to run the command with.
• extension: File extension to be appended to the command to be executed.
• return_all: Whether this task should return all lines of stdout as a list, or just the last line as a string.
• stream_level: The logging level of the stream; defaults to 20 equivalent to logging.INFO.
• cwd: The working directory context the command will be executed within

Returns:

• If return all, returns all lines as a list; else the last line as a string.

Classes

ShellProcess

A class representing a shell process.

Supports both async (anyio.abc.Process) and sync (subprocess.Popen) processes.

Methods:

afetch_result

afetch_result(self) -> list[str]

Retrieve the output of the shell operation (async version).

Returns:

• The lines output from the shell operation as a list.

await_for_completion

await_for_completion(self) -> None

Wait for the shell command to complete after a process is triggered (async version).

fetch_result

fetch_result(self) -> list[str]

Retrieve the output of the shell operation (sync version).

Returns:

• The lines output from the shell operation as a list.

pid

pid(self) -> int

The PID of the process.

Returns:

• The PID of the process.

return_code

return_code(self) -> int | None

The return code of the process.

Returns:

• The return code of the process, or None if the process is still running.

wait_for_completion

wait_for_completion(self) -> None

Wait for the shell command to complete after a process is triggered (sync version).

ShellOperation

A block representing a shell operation, containing multiple commands.

For long-lasting operations, use the trigger method and utilize the block as a context manager for automatic closure of processes when context is exited. If not, manually call the close method to close processes.

For short-lasting operations, use the run method. Context is automatically managed with this method.

Attributes:

• commands: A list of commands to execute sequentially.
• stream_output: Whether to stream output.
• env: A dictionary of environment variables to set for the shell operation.
• working_dir: The working directory context the commands will be executed within.
• shell: The shell to use to execute the commands.
• extension: The extension to use for the temporary file. if unset defaults to .ps1 on Windows and .sh on other platforms.

Examples:

Load a configured block:

from prefect_shell import ShellOperation

shell_operation = ShellOperation.load("BLOCK_NAME")

Methods:

aclose

aclose(self)

Close the job block (async version).

arun

arun(self, **open_kwargs: dict[str, Any]) -> list[str]

Runs a shell command (async version), but unlike the trigger method, additionally waits and fetches the result directly, automatically managing the context. This method is ideal for short-lasting shell commands; for long-lasting shell commands, it is recommended to use the trigger method instead.

Args:

• **open_kwargs: Additional keyword arguments to pass to open_process.

Returns:

• The lines output from the shell command as a list.

Examples:

Sleep for 5 seconds and then print “Hello, world!”:

from prefect_shell import ShellOperation

shell_output = await ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"]
).arun()

atrigger

atrigger(self, **open_kwargs: dict[str, Any]) -> ShellProcess

Triggers a shell command and returns the shell command run object to track the execution of the run (async version). This method is ideal for long-lasting shell commands; for short-lasting shell commands, it is recommended to use the run method instead.

Args:

• **open_kwargs: Additional keyword arguments to pass to open_process.

Returns:

• A ShellProcess object.

Examples:

Sleep for 5 seconds and then print “Hello, world!”:

from prefect_shell import ShellOperation

async with ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"],
) as shell_operation:
    shell_process = await shell_operation.atrigger()
    await shell_process.await_for_completion()
    shell_output = await shell_process.afetch_result()

close

close(self)

Close the job block (sync version).

run

run(self, **open_kwargs: dict[str, Any]) -> list[str]

Runs a shell command (sync version), but unlike the trigger method, additionally waits and fetches the result directly, automatically managing the context. This method is ideal for short-lasting shell commands; for long-lasting shell commands, it is recommended to use the trigger method instead.

Args:

• **open_kwargs: Additional keyword arguments to pass to subprocess.Popen.

Returns:

• The lines output from the shell command as a list.

Examples:

Sleep for 5 seconds and then print “Hello, world!”:

from prefect_shell import ShellOperation

shell_output = ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"]
).run()

trigger

trigger(self, **open_kwargs: dict[str, Any]) -> ShellProcess

Triggers a shell command and returns the shell command run object to track the execution of the run (sync version). This method is ideal for long-lasting shell commands; for short-lasting shell commands, it is recommended to use the run method instead.

Args:

• **open_kwargs: Additional keyword arguments to pass to subprocess.Popen.

Returns:

• A ShellProcess object.

Examples:

Sleep for 5 seconds and then print “Hello, world!”:

from prefect_shell import ShellOperation

with ShellOperation(
    commands=["sleep 5", "echo 'Hello, world!'"],
) as shell_operation:
    shell_process = shell_operation.trigger()
    shell_process.wait_for_completion()
    shell_output = shell_process.fetch_result()