FileSystem Steps

Builtin file system steps.

dir_exists

pyapp_flow.builtin.fs.dir_exists(path: Path | str, output_var='dir_exists') → Step

Check if a directory exists.

Often used with a condition block.

Parameters:

  • path – Path to directory to check for.

  • output_var – Name of output variable.

Returns:

Boolean of directory state.

flow.If(
    dir_exists("/path/to/dir")
).true(...)

file_exists

pyapp_flow.builtin.fs.file_exists(path: Path | str, output_var='file_exists') → Step

Check if a file exists.

Often used with a condition block.

Parameters:

  • path – Path to file to check for.

  • output_var – Name of output variable.

Returns:

Boolean of file state.

flow.If(
    file_exists("/path/to/dir/file.txt")
).true(...)

ensure_dir

pyapp_flow.builtin.fs.ensure_dir(path: Path | str, *, parents: bool = True) → Step

Ensure a directory exists, creating if required.

Parameters:

  • path – Path to file/dir that must exist.

  • parents – If True, create parent directory’s if necessary.

ensure_dir("/path/to/dir")

ensure_parent_dir

pyapp_flow.builtin.fs.ensure_parent_dir(path: Path | str, *, parents: bool = True) → Step

Ensure the parent directory exists, creating if required.

Parameters:

  • path – Path to file/dir whose parent must exist.

  • parents – If True, create parent directory’s if necessary.

ensure_parent_dir("/path/to/dir")

Temp Workspace

class pyapp_flow.builtin.fs.TempWorkspace(target_var: str = 'workspace', *, cleanup: bool = True, prefix: str = 'flow-', base_dir: Path | str | None = None)

Step that creates a temporary workspace that is cleaned up on exit.

Parameters:

  • target_var – Variable the Path to the workspace is placed; default is workspace.

  • cleanup – Clean up the workspace on exit; default is True.

  • prefix – A prefix to provide the workspace name; default is flow-.

  • base_dir – An optional location to create the temporary workspace. The default value of None will resolve the system TEMP variable as the base_dir.

TempWorkspace()
enter(context: WorkflowContext)

Enter the context.

exit(context: WorkflowContext, exception: Exception | None)

Exit the context.

The exception value is supplied if an exception was raised when calling child nodes.

property name: str

Name of the step.