pyantz.jobs.files package

Contents

pyantz.jobs.files package#

Submodules#

pyantz.jobs.files.check module#

Jobs for asserting conditions about files.

file_exists() checks that a path exists; file_contains() uses memory-mapped I/O to efficiently search for a substring anywhere in a file. Both return False (signalling job failure) when the condition is not met.

class ExistsParams(*, file: Path)[source]#

Bases: BaseModel

Parameters needed to check if a file exists.

Parameters:

file (Path)

_abc_impl = <_abc._abc_data object>#
file: Path#

File (or directory) which must exist.

class FileContentsParams(*, file: Path, substring: str)[source]#

Bases: BaseModel

Parameters to check if a file contains a substring.

Parameters:

  • file (Path)

  • substring (str)

_abc_impl = <_abc._abc_data object>#
file: Path#

File to check.

substring: str#

Substring to search for in the file; it may span lines, in which case it must include the line breaks.

file_contains(params: FileContentsParams)[source]#

Check if a file contains a substring.

The file is memory mapped instead of read into memory, so large files can be searched. The substring is encoded as UTF-8 and may span several lines.

Parameters:

params (FileContentsParams) – the job’s parameters, validated as FileContentsParams.

Job parameters:

  • file (Path) – file to check.

  • substring (str) – substring to search for, including any line breaks.

Returns:

True if the file contains the substring, False if it doesn’t or the file doesn’t exist.

Return type:

bool

Raises:

OSError – if the file exists but can’t be read.

Example

from pathlib import Path

from pyantz.jobs.files.check import file_contains

def submit(job):
    pass

Path("log.txt").write_text("step 1: ok\nstep 2: ok\n")
for substring in ("ok\nstep 2", "error"):
    params = {"file": "log.txt", "substring": substring}
    print(file_contains(params, submit))

True
False
file_exists(params: ExistsParams)[source]#

Check if a file exists.

Parameters:

params (ExistsParams) – the job’s parameters, validated as ExistsParams.

Job parameters:

file (Path) – file (or directory) which must exist.

Returns:

True if the path exists, otherwise False.

Return type:

bool

Raises:

None – a missing file is reported by returning False.

Example

from pathlib import Path

from pyantz.jobs.files.check import file_exists

Path("here.txt").touch()
print(file_exists({"file": "here.txt"}, lambda job: None))
print(file_exists({"file": "not_here.txt"}, lambda job: None))

True
False

pyantz.jobs.files.edit module#

In-place editing of JSON and YAML files.

Fields are addressed with dot-separated paths (e.g. "server.port") and array indices in bracket notation (e.g. "items.[0]"). Escape a literal . with a backslash: "a\\.b" refers to the key a.b.

Available jobs:

class EditExternalParams(*, extract_function: _FunctionDefinition[Callable[..., JsonValue]], save_function: _FunctionDefinition[Callable[..., NoneType]], updates: Mapping[str, Any], extra_field: _ExtraField = 'force_object')[source]#

Bases: BaseModel

Edit a file using an external function to extract and to save off.

Parameters:
_abc_impl = <_abc._abc_data object>#
extra_field: _ExtraField#

ignore leaves the value as it was, overwrite replaces the deepest existing leaf, and force_object creates the missing objects.

Type:

What to do when a field doesn’t exist in the file

extract_function: _FunctionDefinition[Callable[..., JsonValue]]#

Function called with its args/kwargs to load the value to edit; it MUST return a JsonValue.

save_function: _FunctionDefinition[Callable[..., NoneType]]#

its FIRST argument is the edited value, followed by its own args/kwargs.

Type:

Function called to save the edited value

updates: Mapping[str, Any]#

Updates to make, mapping each field to its new value.

To edit a subfield, join fields with . (escape a literal . with \\): "a.b" edits the b field of the object in a. To edit arrays, use [] with an index: "a.[0]" edits the first element of a. An index past the end appends, a negative index adds to the front and "[*]" edits every element.

class EditJsonParams(*, file: Path, updates: Mapping[str, Any], extra_field: _ExtraField = 'force_object')[source]#

Bases: _EditJsonLikeParams

Parameters to edit json.

Parameters:

  • file (Path)

  • updates (Mapping[str, Any])

  • extra_field (_ExtraField)

_abc_impl = <_abc._abc_data object>#
class EditYamlParams(*, file: Path, updates: Mapping[str, Any], extra_field: _ExtraField = 'force_object')[source]#

Bases: _EditJsonLikeParams

Parameters to edit a yaml file.

Parameters:

  • file (Path)

  • updates (Mapping[str, Any])

  • extra_field (_ExtraField)

_abc_impl = <_abc._abc_data object>#
JSON_INDENT = 2#

Indentation of the json files which are written.

class _EditJsonLikeParams(*, file: Path, updates: Mapping[str, Any], extra_field: _ExtraField = 'force_object')[source]#

Bases: BaseModel

Parameters to edit a JSON-like input file.

Parameters:

  • file (Path)

  • updates (Mapping[str, Any])

  • extra_field (_ExtraField)

_abc_impl = <_abc._abc_data object>#
extra_field: _ExtraField#

ignore leaves the value as it was, overwrite replaces the deepest existing leaf, and force_object creates the missing objects.

Type:

What to do when a field doesn’t exist in the file

file: Path#

File to edit in place.

updates: Mapping[str, Any]#

Updates to make, mapping each field to its new value.

To edit a subfield, join fields with . (escape a literal . with \\): "a.b" edits the b field of the object in a. To edit arrays, use [] with an index: "a.[0]" edits the first element of a. An index past the end appends, a negative index adds to the front and "[*]" edits every element.

type _ExtraField = Literal['ignore', 'overwrite', 'force_object']#

What to do when an update’s field doesn’t exist in the edited value.

Consider the field "a.b.c.d" being set to "goodbye!" in {"a": "hello"}:

  • ignore: leave the value as it was: {"a": "hello"}.

  • overwrite: follow the field as far as possible, then replace that leaf: {"a": "goodbye!"}.

  • force_object: create the missing objects: {"a": {"b": {"c": {"d": "goodbye!"}}}}.

class _FunctionDefinition[T: Callable[[...], Any]](*, kwargs: dict[str, ~typing.Any]=<factory>, args: tuple[~typing.Any, ...]=<factory>, func: T)[source]#

Bases: BaseModel, Generic[T]

Quick definition of a function to be run.

Parameters:

  • kwargs (dict[str, Any])

  • args (tuple[Any, ...])

  • func (Annotated[T, BeforeValidator(func=~pyantz.infrastructure.config.fn_utils.import_function_by_name, json_schema_input_type=PydanticUndefined), PlainSerializer(func=~pyantz.infrastructure.config.fn_utils.serialize_function, return_type=str, when_used=always), WithJsonSchema(json_schema={'type': 'string', 'format': 'base64'}, mode=None)])

_abc_impl = <_abc._abc_data object>#
args: tuple[Any, ...]#

Positional arguments to call the function with.

func: T#

The function to call, given by its full import path.

kwargs: dict[str, Any]#

Keyword arguments to call the function with.

exception _PathMismatchError[source]#

Bases: Exception

The field being edited does not match the structure of the file.

_edit_jsonlike(params: _EditJsonLikeParams, contents: JsonValue)[source]#

Edit a JSON like value and return the contents.

Parameters:

  • params (_EditJsonLikeParams) – parameters holding the updates to make and how to handle fields which don’t exist.

  • contents (JsonValue) – the value to edit. Mutable mappings are edited in place.

Returns:

the edited value.

Return type:

JsonValue

Raises:

ValueError – if a field is invalid, e.g., it has an empty part ("a..b") or a list index which isn’t an integer.

Example

from pyantz.jobs.files.edit import EditJsonParams, _edit_jsonlike

params = EditJsonParams(
    file="unused.json",
    updates={"a.b": 1},
    extra_field="overwrite",
)
print(_edit_jsonlike(params, {"a": "text"}))

{'a': 1}
_json_editor_factory(extra_field: _ExtraField = 'force_object')[source]#

Create a callable to edit a JsonValue recursively.

The callable takes the value to edit, a field such as "a.[0].b" and the new value, and returns the edited value. Mutable mappings are edited in place (keeping the order of keys and, for yaml, comments), except with ignore, which edits a copy so the original is untouched when a field doesn’t match.

Parameters:

extra_field (_ExtraField) – user has defined a field (nested too deeply); how to handle ignore: don’t overwrite, ignore the user instructions overwrite: follow as far as possible, then overwrite the leaf force_object: turn the value into an object that is as nested as user input

Returns:

the editing function.

Return type:

Callable[[JsonValue, str, JsonValue], JsonValue]

Raises:

None – errors in fields are raised by the returned function, as ValueError.

Example

from pyantz.jobs.files.edit import _json_editor_factory

for extra_field in ("ignore", "overwrite", "force_object"):
    edit = _json_editor_factory(extra_field)
    print(extra_field, edit({"a": "hello"}, "a.b.c", "goodbye!"))

ignore {'a': 'hello'}
overwrite {'a': 'goodbye!'}
force_object {'a': {'b': {'c': 'goodbye!'}}}
_write_atomically(path: Path, write: Callable[[Any], object])[source]#

Write via a temporary file so that a failure leaves the original intact.

The temporary file is created next to path and replaces it only once write has finished.

Parameters:

  • path (Path) – the file to (over)write.

  • write (Callable[[Any], object]) – function which writes the new contents to the open text file it is given.

Return type:

None

Returns:

None.

Raises:

  • OSError – if the temporary file can’t be created or can’t replace path.

  • BaseException – anything raised by write is raised again, once the temporary file has been removed.

Example

from pathlib import Path

from pyantz.jobs.files.edit import _write_atomically

def fail(fh):
    fh.write("half written")
    raise RuntimeError("oops")

Path("notes.txt").write_text("original")
try:
    _write_atomically(Path("notes.txt"), fail)
except RuntimeError:
    pass
print(Path("notes.txt").read_text())

original
edit_external_method(params: EditExternalParams)[source]#

Edit a file using user-provided functionality.

The value to edit is loaded by extract_function, edited like edit_json() would, and given to save_function. This edits formats which have no job of their own.

Parameters:

params (EditExternalParams) – the job’s parameters, validated as EditExternalParams.

Job parameters:

  • extract_function (dict) – func (import path), args and kwargs of the function which returns the value to edit.

  • save_function (dict) – func (import path), args and kwargs of the function which saves the edited value, given as its first argument.

  • updates (Mapping[str, Any]) – new values keyed by their field, such as "a.b" or "items.[0]"; see EditExternalParams.

  • extra_field (str) – what to do when a field doesn’t exist: ignore, overwrite or force_object. Defaults to force_object.

Returns:

True if the value was extracted, edited and saved; False if any of those steps fails.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

from pyantz.jobs.files.edit import edit_external_method

edit_external_method(
    {
        "extract_function": {"func": "json.loads", "args": ['{"a": 1}']},
        "save_function": {"func": "builtins.print"},
        "updates": {"b": 2},
    },
    lambda job: None,
)

{'a': 1, 'b': 2}
edit_json(params: EditJsonParams)[source]#

Edit a .json file in place.

The file is rewritten with an indent of JSON_INDENT. It is written atomically, so a failure leaves the original intact.

Parameters:

params (EditJsonParams) – the job’s parameters, validated as EditJsonParams.

Job parameters:

  • file (Path) – file to edit in place.

  • updates (Mapping[str, Any]) – new values keyed by their field, such as "a.b" or "items.[0]"; see EditJsonParams.

  • extra_field (str) – what to do when a field doesn’t exist: ignore, overwrite or force_object. Defaults to force_object.

Returns:

True if the file was edited, False if it doesn’t exist, isn’t a file, isn’t valid json or a field is invalid.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

import json
from pathlib import Path

from pyantz.jobs.files.edit import edit_json

Path("data.json").write_text('{"items": [{"n": 1}, {"n": 2}]}')
edit_json(
    {
        "file": "data.json",
        "updates": {"items.[*].n": 0, "items.[5]": "appended"},
    },
    lambda job: None,
)
print(json.loads(Path("data.json").read_text()))

{'items': [{'n': 0}, {'n': 0}, 'appended']}
edit_yaml(params: EditYamlParams)[source]#

Edit a .yaml file in place.

The file is edited with ruamel.yaml, which keeps its comments and the order of its keys. It is written atomically, so a failure leaves the original intact.

Parameters:

params (EditYamlParams) – the job’s parameters, validated as EditYamlParams.

Job parameters:

  • file (Path) – file to edit in place.

  • updates (Mapping[str, Any]) – new values keyed by their field, such as "a.b" or "items.[0]"; see EditJsonParams.

  • extra_field (str) – what to do when a field doesn’t exist: ignore, overwrite or force_object. Defaults to force_object.

Returns:

True if the file was edited, False if it doesn’t exist, isn’t a file, can’t be parsed or a field is invalid.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

from pathlib import Path

from pyantz.jobs.files.edit import edit_yaml

Path("config.yaml").write_text("server:\n  port: 80  # the port\n")
edit_yaml(
    {"file": "config.yaml", "updates": {"server.port": 8080}},
    lambda job: None,
)
print(Path("config.yaml").read_text(), end="")

server:
  port: 8080 # the port

pyantz.jobs.files.moving module#

File and directory transfer operations.

Provides rsync() (shell-based, ideal for remote or large trees) and pure-Python copy() / move() alternatives built on shutil. All three log errors and return False on failure rather than raising exceptions.

class CopyParams(*, source: Path | str, destination: Path | str, overwrite: bool = True)[source]#

Bases: BaseModel

Copy parameters to copy from location to location.

Parameters:

  • source (Path | str)

  • destination (Path | str)

  • overwrite (bool)

_abc_impl = <_abc._abc_data object>#
destination: Path | str#

Destination directory or file.

overwrite: bool#

If true, will overwrite the destination.

source: Path | str#

Source directory or file.

class MoveParams(*, source: Path, destination: Path, overwrite: bool = True)[source]#

Bases: BaseModel

Parameters to move a file or directory.

Parameters:

  • source (Path)

  • destination (Path)

  • overwrite (bool)

_abc_impl = <_abc._abc_data object>#
destination: Path#

Destination directory or file.

overwrite: bool#

If true, will overwrite the destination.

source: Path#

Source directory or file.

class RsyncParams(*, source: Path, destination: Path)[source]#

Bases: BaseModel

Rsync parameters to copy from location to location.

Parameters:

  • source (Path)

  • destination (Path)

_abc_impl = <_abc._abc_data object>#
destination: Path#

Destination directory or file.

source: Path#

Source directory or file.

_remove(path: Path)[source]#

Delete a file or directory tree if it exists.

Parameters:

path (Path) – the file, symbolic link or directory to delete.

Return type:

None

Returns:

None.

Raises:

OSError – if the path exists but can’t be deleted.

Example

from pathlib import Path

from pyantz.jobs.files.moving import _remove

Path("tree/inner").mkdir(parents=True)
_remove(Path("tree"))
_remove(Path("never_existed.txt"))  # missing paths are fine
print(Path("tree").exists())

False
copy(params: CopyParams)[source]#

Use python shutil to copy directories and files.

The copy is made next to the destination first, and only replaces an existing destination once it is complete, so a failed copy leaves the destination as it was.

Parameters:

params (CopyParams) – the job’s parameters, validated as CopyParams.

Job parameters:

  • source (Path | str) – source directory or file.

  • destination (Path | str) – destination directory or file.

  • overwrite (bool) – if True, replace an existing destination. Defaults to True.

Returns:

True if copied, False if the source doesn’t exist, the destination exists and overwrite is False, or the copy fails.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

from pathlib import Path

from pyantz.jobs.files.moving import copy

def submit(job):
    pass

Path("new.txt").write_text("new")
Path("old.txt").write_text("old")
params = {"source": "new.txt", "destination": "old.txt"}
print(copy({**params, "overwrite": False}, submit))
print(copy(params, submit), Path("old.txt").read_text())

False
True new
move(params: MoveParams)[source]#

Move from a source to a destination.

Uses shutil.move(), so the move may cross file systems. An existing destination is kept aside until the move has worked, and restored if it fails.

Parameters:

params (MoveParams) – the job’s parameters, validated as MoveParams.

Job parameters:

  • source (Path) – source directory or file.

  • destination (Path) – destination directory or file.

  • overwrite (bool) – if True, replace an existing destination. Defaults to True.

Returns:

True if moved, False if the source doesn’t exist, the destination exists and overwrite is False, or the move fails.

Return type:

bool

Raises:

OSError – if an existing destination can’t be set aside before the move.

Example

from pathlib import Path

from pyantz.jobs.files.moving import move

Path("results").mkdir()
Path("results/out.txt").write_text("done")
move({"source": "results", "destination": "archive"}, lambda job: None)
print(Path("results").exists(), Path("archive/out.txt").read_text())

False done
rsync(params: RsyncParams)[source]#

Use rsync to copy from source to destination.

Runs rsync -av source destination, so the rsync program must be installed. As with rsync itself, a trailing / on a source directory copies its contents instead of the directory.

Parameters:

params (RsyncParams) – the job’s parameters, validated as RsyncParams.

Job parameters:

  • source (Path) – source directory or file.

  • destination (Path) – destination directory or file.

Returns:

True if rsync succeeded, False if the source doesn’t exist, rsync can’t be run or rsync fails.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

import shutil
from pathlib import Path

from pyantz.jobs.files.moving import rsync

Path("data.txt").write_text("hello")
if shutil.which("rsync"):
    rsync({"source": "data.txt", "destination": "synced.txt"}, None)
else:  # not installed here, so do what rsync would have done
    shutil.copy("data.txt", "synced.txt")
print(Path("synced.txt").read_text())

hello

pyantz.jobs.files.reformat module#

Reformat between tabular datasets.

class FromCsvToParquetParameters(*, csv_file: Path, parquet_file: Path)[source]#

Bases: BaseModel

Load a csv and dump it into a parquet.

Parameters:

  • csv_file (Path)

  • parquet_file (Path)

_abc_impl = <_abc._abc_data object>#
csv_file: Path#

Csv file to read.

parquet_file: Path#

Parquet file to write.

class FromExcelToCsvParameters(*, excel_file: Path, sheet_name: str | None = None, csv_file: Path)[source]#

Bases: BaseModel

Load a page from an excel and dump it into a csv.

Parameters:

  • excel_file (Path)

  • sheet_name (str | None)

  • csv_file (Path)

_abc_impl = <_abc._abc_data object>#
csv_file: Path#

Csv file to write.

excel_file: Path#

Excel workbook to read.

sheet_name: str | None#

Name of the sheet to convert; if None, the first sheet.

csv_to_parquet(params: FromCsvToParquetParameters)[source]#

Use polars to load a csv and dump it into a parquet.

The csv is streamed into the parquet, so it doesn’t need to fit in memory.

Parameters:

params (FromCsvToParquetParameters) – the job’s parameters, validated as FromCsvToParquetParameters.

Job parameters:

  • csv_file (Path) – csv file to read.

  • parquet_file (Path) – parquet file to write.

Returns:

True if the parquet was written, False if the csv doesn’t exist or can’t be converted.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

import polars as pl
from polars.testing import assert_frame_equal

from pyantz import start

start({
    "submitter": {
        "type_": "local_proc",
        "working_directory": ".",
    },
    "jobs": [
        {
            "function": "pyantz.jobs.files.reformat.csv_to_parquet",
            "parameters": {
                "csv_file": "some_csv.csv",
                "parquet_file": "reformatted.parquet",
            },
        },
    ],
})

from_csv = pl.read_csv("some_csv.csv")
from_parq = pl.read_parquet("reformatted.parquet")

assert_frame_equal(from_csv, from_parq)
excel_to_csv(params: FromExcelToCsvParameters)[source]#

Use polars to load an excel and dump it into a csv.

Reading excel files needs the fastexcel package (pip install pyantz[excel]).

Parameters:

params (FromExcelToCsvParameters) – the job’s parameters, validated as FromExcelToCsvParameters.

Job parameters:

  • excel_file (Path) – excel workbook to read.

  • sheet_name (str | None) – name of the sheet to convert. Defaults to None (the first sheet).

  • csv_file (Path) – csv file to write.

Returns:

True if the csv was written, False if the workbook doesn’t exist, fastexcel isn’t installed or the conversion fails.

Return type:

bool

Raises:

None – failures are logged and reported by returning False.

Example

import polars as pl
from polars.testing import assert_frame_equal

from pyantz import start

start({
    "submitter": {
        "type_": "local_proc",
        "working_directory": ".",
    },
    "jobs": [
        {
            "function": "pyantz.jobs.files.reformat.excel_to_csv",
            "parameters": {
                "excel_file": "some_excel.xlsx",
                "csv_file": "some_csv.csv",
            },
        },
    ],
})

from_csv = pl.read_excel("some_excel.xlsx")
from_parq = pl.read_csv("some_csv.csv")

assert_frame_equal(from_csv, from_parq)

pyantz.jobs.files.simple module#

Basic file-system CRUD operations.

Each function is decorated with add_parameters() so it integrates with PyAntz’s type-checked parameter system.

Available jobs:

  • copy() — copy a file or directory tree.

  • move() — rename/move a path.

  • delete() — delete a file or directory.

  • create() — create a file with optional contents.

  • mkdir() — create a directory (including parents).

class CopyParams(*, src: Path, dst: Path)[source]#

Bases: BaseModel

Parameters to copy file.

Parameters:

  • src (Path)

  • dst (Path)

_abc_impl = <_abc._abc_data object>#
dst: Path#

Location to create the copy.

src: Path#

Source file or directory to copy.

class CreateFileParams(*, path: Path, contents: str | None = None, overwrite: bool = True)[source]#

Bases: BaseModel

Required inputs to create a file.

Parameters:

  • path (Path)

  • contents (str | None)

  • overwrite (bool)

_abc_impl = <_abc._abc_data object>#
contents: str | None#

Optionally, text to write to the file.

overwrite: bool#

If true, replaces the file if it exists (if contents is empty, clears it).

path: Path#

Path to the file to create.

class DeleteParams(*, path: Path, missing_ok: bool = True, recursive: bool = False)[source]#

Bases: BaseModel

Parameters to define what to delete.

Parameters:

  • path (Path)

  • missing_ok (bool)

  • recursive (bool)

_abc_impl = <_abc._abc_data object>#
missing_ok: bool#

If true, deleting a path which doesn’t exist succeeds.

path: Path#

Location to delete.

recursive: bool#

If true, directories are deleted with everything inside them (otherwise only empty directories can be deleted).

class Mkdirparams(*, dir_path: Path)[source]#

Bases: BaseModel

Required input to create a directory.

Parameters:

dir_path (Path)

_abc_impl = <_abc._abc_data object>#
dir_path: Path#

Path of the directory to create.

class MoveParams(*, src: Path, dst: Path)[source]#

Bases: BaseModel

Parameters to move a file.

Parameters:

  • src (Path)

  • dst (Path)

_abc_impl = <_abc._abc_data object>#
dst: Path#

Location to place the result.

src: Path#

Source file or directory to move.

copy(params: CopyParams)[source]#

Copy from src to destination.

A file is copied with its metadata (shutil.copy2()); a directory is copied with everything inside it (shutil.copytree()).

Parameters:

params (CopyParams) – the job’s parameters, validated as CopyParams.

Job parameters:

  • src (Path) – source file or directory to copy.

  • dst (Path) – location to create the copy.

Returns:

True once the copy is made.

Return type:

bool

Raises:

OSError – if the copy fails, e.g., src doesn’t exist or dst is a directory which already exists.

Example

from pathlib import Path

from pyantz.jobs.files.simple import copy

Path("original.txt").write_text("hello")
copy({"src": "original.txt", "dst": "copied.txt"}, lambda job: None)
print(Path("copied.txt").read_text())

hello
create(params: CreateFileParams)[source]#

Create the file as requested and write contents if provided.

If the file exists and overwrite is false, its contents are never changed: asking to write contents to it fails.

Parameters:

params (CreateFileParams) – the job’s parameters, validated as CreateFileParams.

Job parameters:

  • path (Path) – path to the file to create.

  • contents (str | None) – text to write to the file. Defaults to None (an empty file).

  • overwrite (bool) – if True, replace an existing file. Defaults to True.

Returns:

True if the file was created, False if it exists, overwrite is False and contents were given.

Return type:

bool

Raises:

OSError – if the file can’t be written, e.g., its directory doesn’t exist.

Example

from pathlib import Path

from pyantz.jobs.files.simple import create

def submit(job):
    pass

print(create({"path": "new.txt", "contents": "first"}, submit))
print(
    create(
        {"path": "new.txt", "contents": "second", "overwrite": False},
        submit,
    )
)
print(Path("new.txt").read_text())

True
False
first
delete(params: DeleteParams)[source]#

Delete the file passed in delete params.

Parameters:

params (DeleteParams) – the job’s parameters, validated as DeleteParams.

Job parameters:

  • path (Path) – file or directory to delete.

  • missing_ok (bool) – if True, a missing path counts as success. Defaults to True.

  • recursive (bool) – if True, delete directories with everything inside them. Defaults to False.

Returns:

True if the path was deleted (or was missing and missing_ok), False if it was missing and not missing_ok, or is a directory which isn’t empty and not recursive.

Return type:

bool

Raises:

OSError – if a file or tree can’t be deleted, e.g., for lack of permission.

Example

from pathlib import Path

from pyantz.jobs.files.simple import delete

def submit(job):
    pass

Path("full_dir").mkdir()
Path("full_dir/file.txt").touch()
print(delete({"path": "full_dir"}, submit))
print(delete({"path": "full_dir", "recursive": True}, submit))
print(delete({"path": "full_dir", "missing_ok": False}, submit))

False
True
False
mkdir(params: Mkdirparams)[source]#

Create the directory, and any missing parents (it may already exist).

Parameters:

params (Mkdirparams) – the job’s parameters, validated as Mkdirparams.

Job parameters:

dir_path (Path) – path of the directory to create.

Returns:

True once the directory exists.

Return type:

bool

Raises:

OSError – if the directory can’t be created, e.g., a file has its path.

Example

from pathlib import Path

from pyantz.jobs.files.simple import mkdir

mkdir({"dir_path": "a/b/c"}, lambda job: None)
print(Path("a/b/c").is_dir())

True
move(params: MoveParams)[source]#

Move src to dst.

Uses pathlib.Path.rename(), so both paths must be on the same file system; see pyantz.jobs.files.moving.move() for moves across them.

Parameters:

params (MoveParams) – the job’s parameters, validated as MoveParams.

Job parameters:

  • src (Path) – source file or directory to move.

  • dst (Path) – location to place the result.

Returns:

True once moved.

Return type:

bool

Raises:

OSError – if the move fails, e.g., src doesn’t exist.

Example

from pathlib import Path

from pyantz.jobs.files.simple import move

Path("before.txt").write_text("hello")
move({"src": "before.txt", "dst": "after.txt"}, lambda job: None)
print(Path("before.txt").exists(), Path("after.txt").read_text())

False hello

Module contents#

Common file operations.

Contents