ddp_utils.file_manager¶

import ddp_utils.file_manager

Provide high-level file operations backed by PathInfo.

The module supports text, JSON, JSONL, and CSV operations, including atomic writes, retries, streaming, file watching, checksums, and gzip compression. Attributes not implemented by FileManager are delegated to its underlying PathInfo instance.

Examples

Write and read a text file:

from ddp_utils.file_manager import FileManager

file = FileManager("runtime/state.txt")
file.overwrite("ready")
assert file.read() == "ready"

class ddp_utils.file_manager.FileManager(file_path: str | PathLike | Path)¶

Bases: object

Manage structured and unstructured files through one facade.

High-level read and write operations live on this class, while low-level path operations are delegated to path_info. Write operations create missing parent directories automatically and store the most recent caught exception in last_error.

Parameters

Name	Type	Description
file_path	str \| PathLike \| Path	Target file path accepted by `PathInfo`.

Examples

Create a manager for a nested runtime file:

file = FileManager("runtime/results.json")

Initialize a manager for one target path.

Parameters

Name	Type	Description
file_path	str \| PathLike \| Path	Target file path.

Examples

Wrap a path-like value:

file = FileManager(Path("runtime/output.txt"))

property last_error: Exception | None¶

Return the most recent caught file-operation exception.

Returns

The exception object, or None before an error or after success.

Examples

Inspect a failed read without raising:

file = FileManager("missing.txt")
file.read()
error = file.last_error

property last_error_message: str | None¶

Return text for the most recent caught exception.

Returns

str(last_error), or None when no error is stored.

Examples

Log a failure without handling the exception object:

message = FileManager("missing.txt").last_error_message

create(content: Any = '', encoding: str = 'utf-8') → None¶

Create or truncate the target and write its initial content.

Dictionaries and lists are serialized as indented JSON. Other values are converted with str.

Parameters

Name	Type	Description
content	Any	Initial text, JSON mapping, JSON list, or stringable value.
encoding	str	Text encoding used to open the target.

Examples

Create a JSON file and its missing parent directories:

FileManager("runtime/state.json").create({"ready": True})

create_retry(content: Any = '', encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry create() after transient failures.

Parameters

Name	Type	Description
content	Any	Initial content passed to `create()`.
encoding	str	Target text encoding.
attempts	int	Total write attempts, including the first.
delay	float	Initial delay between failed attempts.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional maximum delay between attempts.

Returns

Type	Description
bool	`True` after a successful write; otherwise `False`.

Examples

Retry creation on a synchronized filesystem:

created = FileManager("shared/state.txt").create_retry(
    "ready", attempts=5
)

create_atomic(content: Any = '', encoding: str = 'utf-8') → str | PathLike | Path | None¶

Create or replace the target through an atomic file swap.

Dictionaries and lists are serialized as indented JSON. Other values are converted with str.

Parameters

Name	Type	Description
content	Any	Text, JSON mapping, JSON list, or stringable value.
encoding	str	Target text encoding.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None` with `last_error` populated.

Examples

Atomically replace a state document:

path = FileManager("state.json").create_atomic({"step": 2})

create_atomic_retry(content: Any = '', encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → str | PathLike | Path | None¶

Retry create_atomic() after transient failures.

Parameters

Name	Type	Description
content	Any	Content passed to `create_atomic()`.
encoding	str	Target text encoding.
attempts	int	Total write attempts, including the first.
delay	float	Initial delay between failed attempts.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional maximum delay between attempts.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise the final `None` result.

Examples

Retry an atomic state write:

path = FileManager("state.json").create_atomic_retry(
    {"step": 2}, attempts=5
)

overwrite(content: str, encoding: str = 'utf-8') → None¶

Replace the target contents with text.

Parameters

Name	Type	Description
content	str	Complete replacement text.
encoding	str	Target text encoding.

Examples

Replace an existing status file:

FileManager("status.txt").overwrite("ready")

overwrite_retry(content: str, encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry overwrite() after transient failures.

Parameters

Name	Type	Description
content	str	Complete replacement text.
encoding	str	Target text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful overwrite; otherwise `False`.

Examples

Retry a write on a synchronized directory:

ok = FileManager("shared/status.txt").overwrite_retry(
    "ready", attempts=5
)

overwrite_atomic(content: str, encoding: str = 'utf-8') → str | PathLike | Path | None¶

Atomically replace the target contents with text.

Parameters

Name	Type	Description
content	str	Complete replacement text.
encoding	str	Target text encoding.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Examples

Replace a checkpoint without exposing a partial file:

path = FileManager("checkpoint.txt").overwrite_atomic("step=2")

overwrite_atomic_retry(content: str, encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → str | PathLike | Path | None¶

Retry overwrite_atomic() after transient failures.

Parameters

Name	Type	Description
content	str	Complete replacement text.
encoding	str	Target text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry an atomic checkpoint replacement:

path = FileManager("checkpoint.txt").overwrite_atomic_retry(
    "step=2", attempts=5
)

append(content: str, encoding: str = 'utf-8', newline: bool = True) → bool¶

Append text directly to the target.

Parameters

Name	Type	Description
content	str	Text to append.
encoding	str	Target text encoding.
newline	bool	Whether to append `\n` after the text.

Returns

Type	Description
bool	`True` on success; otherwise `False` with the error stored.

Examples

Append one line to a log:

ok = FileManager("events.log").append("started")

append_retry(content: str, encoding: str = 'utf-8', newline: bool = True, *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry append() after transient failures.

Parameters

Name	Type	Description
content	str	Text to append.
encoding	str	Target text encoding.
newline	bool	Whether to append `\n`.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful append; otherwise `False`.

Examples

Retry a log append:

ok = FileManager("events.log").append_retry("started", attempts=5)

append_atomic(content: str, encoding: str = 'utf-8', newline: bool = True) → bool¶

Append text by locking, rebuilding, and atomically replacing the file.

A cross-process FileLock protects the read-modify-replace sequence. This operation reads the complete existing file into memory.

Parameters

Name	Type	Description
content	str	Text to append.
encoding	str	Target text encoding.
newline	bool	Whether to append `\n` after the text.

Returns

Type	Description
bool	`True` after replacement; otherwise `False`.

Examples

Safely append a small checkpoint line across processes:

ok = FileManager("checkpoint.log").append_atomic("step=2")

append_atomic_retry(content: str, encoding: str = 'utf-8', newline: bool = True, *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry append_atomic() after transient failures.

Parameters

Name	Type	Description
content	str	Text to append.
encoding	str	Target text encoding.
newline	bool	Whether to append `\n`.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful append; otherwise `False`.

Examples

Retry a locked atomic append:

ok = FileManager("checkpoint.log").append_atomic_retry(
    "step=2", attempts=5
)

read(encoding: str = 'utf-8') → str | None¶

Read the complete target as text.

Parameters

Name	Type	Description
encoding	str	Source text encoding.

Returns

Type	Description
str \| None	File contents, or `None` when the target is unavailable or an error occurs.

Examples

Read a UTF-8 status file:

status = FileManager("status.txt").read()

lines(encoding: str = 'utf-8') → List[str]¶

Read the target into a list of lines.

Parameters

Name	Type	Description
encoding	str	Source text encoding.

Returns

Type	Description
List[str]	Lines including their retained line endings, or an empty list when the target is unavailable or an error occurs.

Examples

Iterate over the lines of a small text file:

for line in FileManager("events.log").lines():
    print(line.rstrip())

read_json(any_file: bool = False, encoding: str = 'utf-8') → dict | list | None¶

Deserialize a JSON mapping or list from the target.

Parameters

Name	Type	Description
any_file	bool	Whether to accept a target whose suffix is not `.json`.
encoding	str	Source text encoding.

Returns

Type	Description
dict \| list \| None	The decoded mapping or list, or `None` when the file is missing, the suffix is rejected, or decoding fails.

Examples

Load a JSON state mapping:

state = FileManager("state.json").read_json()

write_json(data: dict | list, encoding: str = 'utf-8') → str | PathLike | Path | None¶

Serialize a mapping or list by directly replacing the target.

Parameters

Name	Type	Description
data	dict \| list	JSON-compatible mapping or list.
encoding	str	Target text encoding.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Examples

Write an indented JSON state file:

path = FileManager("state.json").write_json({"step": 2})

Retry write_json() after transient failures.

Parameters

Name	Type	Description
data	dict \| list	JSON-compatible mapping or list.
encoding	str	Target text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry a direct JSON write:

path = FileManager("state.json").write_json_retry(
    {"step": 2}, attempts=5
)

write_json_atomic(data: dict | list, encoding: str = 'utf-8') → str | PathLike | Path | None¶

Serialize JSON through an atomic file replacement.

Parameters

Name	Type	Description
data	dict \| list	JSON-compatible mapping or list.
encoding	str	Target text encoding.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Examples

Atomically publish a JSON checkpoint:

path = FileManager("state.json").write_json_atomic({"step": 2})

Retry write_json_atomic() after transient failures.

Parameters

Name	Type	Description
data	dict \| list	JSON-compatible mapping or list.
encoding	str	Target text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry an atomic JSON checkpoint write:

path = FileManager("state.json").write_json_atomic_retry(
    {"step": 2}, attempts=5
)

append_jsonl(data: Any, encoding: str = 'utf-8', *, ensure_ascii: bool = False, sort_keys: bool = False, compact: bool = True, default: Callable[[Any], Any] | None = None, fsync: bool = True) → bool¶

Append one serialized value as one JSONL record.

The method opens the file with O_APPEND, writes the complete encoded record, and optionally calls os.fsync(). It does not read or rebuild existing records.

Parameters

Name	Type	Description
data	Any	JSON-serializable record.
encoding	str	File encoding.
ensure_ascii	bool	Whether to escape non-ASCII characters.
sort_keys	bool	Whether to sort mapping keys.
compact	bool	Whether to omit optional JSON whitespace.
default	Callable[[Any], Any] \| None	Optional serializer for unsupported values.
fsync	bool	Whether to synchronize the descriptor after the record.

Returns

Type	Description
bool	`True` on success; otherwise `False` with `last_error` populated.

Examples

Persist one processing result immediately:

ok = FileManager("results.jsonl").append_jsonl({"id": 1})

append_jsonl_retry(data: Any, encoding: str = 'utf-8', *, ensure_ascii: bool = False, sort_keys: bool = False, compact: bool = True, default: Callable[[Any], Any] | None = None, fsync: bool = True, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry append_jsonl() after transient failures.

Parameters

Name	Type	Description
data	Any	JSON-serializable record.
encoding	str	File encoding.
ensure_ascii	bool	Whether to escape non-ASCII characters.
sort_keys	bool	Whether to sort mapping keys.
compact	bool	Whether to omit optional JSON whitespace.
default	Callable[[Any], Any] \| None	Optional serializer for unsupported values.
fsync	bool	Whether to synchronize after every attempt that writes.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful append; otherwise `False`.

Examples

Retry persistence on a synchronized directory:

ok = FileManager("results.jsonl").append_jsonl_retry(
    {"id": 1}, attempts=5
)

write_jsonl_atomic(rows: List[Any], encoding: str = 'utf-8', *, ensure_ascii: bool = False, sort_keys: bool = False, compact: bool = True, default: Callable[[Any], Any] | None = None, fail_if_empty: bool = False) → str | PathLike | Path | None¶

Atomically replace the complete JSONL file from a list of records.

This is a full-file temporary write followed by os.replace(); use append_jsonl() for incremental persistence of large files.

Parameters

Name	Type	Description
rows	List[Any]	JSON-serializable records.
encoding	str	File encoding.
ensure_ascii	bool	Whether to escape non-ASCII characters.
sort_keys	bool	Whether to sort mapping keys.
compact	bool	Whether to omit optional JSON whitespace.
default	Callable[[Any], Any] \| None	Optional serializer for unsupported values.
fail_if_empty	bool	Whether an empty list should cancel the write.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Raises

Exception	Description
ValueError	If there are no rows and `fail_if_empty` is true; the write is cancelled.

Examples

Publish a completed JSONL dataset:

path = FileManager("results.jsonl").write_jsonl_atomic(
    [{"id": 1}, {"id": 2}]
)

write_jsonl_atomic_retry(rows: List[Any], encoding: str = 'utf-8', *, ensure_ascii: bool = False, sort_keys: bool = False, compact: bool = True, default: Callable[[Any], Any] | None = None, fail_if_empty: bool = False, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → str | PathLike | Path | None¶

Retry write_jsonl_atomic() after transient failures.

Parameters

Name	Type	Description
rows	List[Any]	JSON-serializable records.
encoding	str	File encoding.
ensure_ascii	bool	Whether to escape non-ASCII characters.
sort_keys	bool	Whether to sort mapping keys.
compact	bool	Whether to omit optional JSON whitespace.
default	Callable[[Any], Any] \| None	Optional serializer for unsupported values.
fail_if_empty	bool	Whether an empty list should cancel the write.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry publication of a completed JSONL dataset:

path = FileManager("results.jsonl").write_jsonl_atomic_retry(
    [{"id": 1}], attempts=5
)

iter_jsonl(encoding: str = 'utf-8', *, skip_empty: bool = True, strict: bool = True) → Iterator[Any]¶

Yield decoded JSONL records without loading the entire file.

Parameters

Name	Type	Description
encoding	str	File encoding.
skip_empty	bool	Whether to ignore empty lines.
strict	bool	Whether invalid or disallowed empty lines should stop iteration. Non-strict mode skips them and records the error.

Yields

The decoded Python value for each accepted record.

Raises

Exception	Description
json.JSONDecodeError	Strict parsing encounters invalid JSON or a disallowed empty line.
OSError	The source cannot be read.

Examples

Process a large result stream record by record:

for record in FileManager("results.jsonl").iter_jsonl():
    process(record)

read_jsonl(encoding: str = 'utf-8', *, skip_empty: bool = True, strict: bool = True) → List[Any]¶

Read all JSONL records into memory.

Prefer iter_jsonl() when the complete decoded list may be large.

Parameters

Name	Type	Description
encoding	str	File encoding.
skip_empty	bool	Whether to ignore empty lines.
strict	bool	Whether an invalid line should fail the complete read.

Returns

Type	Description
List[Any]	Decoded records, or an empty list after an error. The exception is available through `last_error`.

Examples

Load a small result file:

records = FileManager("results.jsonl").read_jsonl()

Convert JSONL records to one atomically published JSON document.

Without root_key the output is a JSON array. With root_key the array is stored under that top-level mapping key.

Parameters

Name	Type	Description
output_path	str \| PathLike \| Path	Final JSON target path.
encoding	str	Input and output text encoding.
root_key	str \| None	Optional key wrapping the record list.
skip_empty	bool	Whether to ignore empty source lines.
strict	bool	Whether an invalid source line should cancel conversion.
indent	int	Output JSON indentation width.
ensure_ascii	bool	Whether to escape non-ASCII output characters.

Returns

Type	Description
str \| PathLike \| Path \| None	The output path on success; otherwise `None`.

Examples

Wrap final records under a documents key:

path = FileManager("results.jsonl").jsonl_to_json_atomic(
    "final.json", root_key="documents"
)

read_csv(delimiter: str = ',', encoding: str = 'utf-8') → List[dict]¶

Read a .csv target into row mappings.

Parameters

Name	Type	Description
delimiter	str	Field separator.
encoding	str	Source text encoding.

Returns

Type	Description
List[dict]	Decoded rows, or an empty list when the suffix is not `.csv`, the target is unavailable, or reading fails.

Examples

Read comma-separated rows:

rows = FileManager("results.csv").read_csv()

append_csv(row: dict, delimiter: str = ',', encoding: str = 'utf-8') → bool¶

Append one CSV row, creating a header for an empty target.

Parameters

Name	Type	Description
row	dict	Mapping whose keys define the fields for this write.
delimiter	str	Field separator.
encoding	str	Target text encoding.

Returns

Type	Description
bool	`True` on success; otherwise `False`.

Examples

Append one result row:

ok = FileManager("results.csv").append_csv({"id": 1, "ok": True})

append_csv_retry(row: dict, delimiter: str = ',', encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry append_csv() after transient failures.

Parameters

Name	Type	Description
row	dict	Mapping to append.
delimiter	str	Field separator.
encoding	str	Target text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful append; otherwise `False`.

Examples

Retry appending a shared CSV file:

ok = FileManager("results.csv").append_csv_retry(
    {"id": 1}, attempts=5
)

append_csv_atomic(row: dict, delimiter: str = ',', encoding: str = 'utf-8') → bool¶

Append a row by atomically rebuilding the complete CSV file.

Existing rows are loaded into memory. The output header is the ordered union of existing fields and keys from the new row.

Parameters

Name	Type	Description
row	dict	Mapping to append.
delimiter	str	Field separator.
encoding	str	Input and output text encoding.

Returns

Type	Description
bool	`True` after the atomic replacement; otherwise `False`.

Examples

Atomically add a field-bearing row:

ok = FileManager("results.csv").append_csv_atomic({"id": 2})

append_csv_atomic_retry(row: dict, delimiter: str = ',', encoding: str = 'utf-8', *, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0) → bool¶

Retry append_csv_atomic() after transient failures.

Parameters

Name	Type	Description
row	dict	Mapping to append.
delimiter	str	Field separator.
encoding	str	Input and output text encoding.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.

Returns

Type	Description
bool	`True` after a successful atomic append; otherwise `False`.

Examples

Retry rebuilding a shared CSV file:

ok = FileManager("results.csv").append_csv_atomic_retry(
    {"id": 2}, attempts=5
)

write_csv(data: List[Dict[str, Any]], *, delimiter: str = ',', encoding: str = 'utf-8-sig', include_header: bool = True, headers: List[str] | None = None, mode: str = 'w', quoting: int = 0, lineterminator: str = '\r\n', extrasaction: Literal['raise', 'ignore'] = 'ignore', sort_keys: bool = False, fail_if_empty: bool = False, **kwargs) → str | PathLike | Path | None¶

Write row mappings directly to the target CSV file.

Parameters

Name	Type	Description
data	List[Dict[str, Any]]	Row mappings to write.
delimiter	str	Field separator.
encoding	str	Target text encoding.
include_header	bool	Whether to write the field names.
headers	List[str] \| None	Explicit field order, or `None` to use the first row.
mode	str	Text open mode, normally `w` or `a`.
quoting	int	`csv` quoting mode.
lineterminator	str	Row terminator.
extrasaction	Literal['raise', 'ignore']	Handling for row keys absent from `headers`.
sort_keys	bool	Whether to sort the selected headers.
fail_if_empty	bool	Whether empty `data` should cancel the write.
**kwargs		Additional arguments passed to `csv.DictWriter`.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Raises

Exception	Description
ValueError	Internally raised for forbidden empty data or missing headers, then captured in `last_error` and returned as `None`.

Examples

Write two rows with inferred headers:

path = FileManager("results.csv").write_csv(
    [{"id": 1}, {"id": 2}]
)

write_csv_retry(data: List[Dict[str, Any]], *, delimiter: str = ',', encoding: str = 'utf-8-sig', include_header: bool = True, headers: List[str] | None = None, mode: str = 'w', quoting: int = 0, lineterminator: str = '\r\n', extrasaction: Literal['raise', 'ignore'] = 'ignore', sort_keys: bool = False, fail_if_empty: bool = False, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0, **kwargs) → str | PathLike | Path | None¶

Retry write_csv() after transient failures.

Parameters

Name	Type	Description
data	List[Dict[str, Any]]	Row mappings to write.
delimiter	str	Field separator.
encoding	str	Target text encoding.
include_header	bool	Whether to write headers.
headers	List[str] \| None	Explicit field order.
mode	str	Target open mode.
quoting	int	`csv` quoting mode.
lineterminator	str	Row terminator.
extrasaction	Literal['raise', 'ignore']	Handling for extra row keys.
sort_keys	bool	Whether to sort headers.
fail_if_empty	bool	Whether empty data should fail.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.
**kwargs		Additional `csv.DictWriter` arguments.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry writing a shared CSV file:

path = FileManager("results.csv").write_csv_retry(
    [{"id": 1}], attempts=5
)

write_csv_atomic(data: List[Dict[str, Any]], *, delimiter: str = ',', encoding: str = 'utf-8-sig', include_header: bool = True, headers: List[str] | None = None, quoting: int = 0, lineterminator: str = '\r\n', extrasaction: Literal['raise', 'ignore'] = 'ignore', sort_keys: bool = False, fail_if_empty: bool = False, **kwargs) → str | PathLike | Path | None¶

Atomically replace the complete target CSV file.

Parameters

Name	Type	Description
data	List[Dict[str, Any]]	Row mappings to write.
delimiter	str	Field separator.
encoding	str	Target text encoding.
include_header	bool	Whether to write headers.
headers	List[str] \| None	Explicit field order, or `None` to infer it.
quoting	int	`csv` quoting mode.
lineterminator	str	Row terminator.
extrasaction	Literal['raise', 'ignore']	Handling for extra row keys.
sort_keys	bool	Whether to sort headers.
fail_if_empty	bool	Whether empty data should cancel the write.
**kwargs		Additional `csv.DictWriter` arguments.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path on success; otherwise `None`.

Raises

Exception	Description
ValueError	If the data is empty and `fail_if_empty` is true, or if the headers cannot be determined because both `data` and `headers` are empty.

Examples

Publish a complete CSV dataset atomically:

path = FileManager("results.csv").write_csv_atomic([{"id": 1}])

write_csv_atomic_retry(data: List[Dict[str, Any]], *, delimiter: str = ',', encoding: str = 'utf-8-sig', include_header: bool = True, headers: List[str] | None = None, quoting: int = 0, lineterminator: str = '\r\n', extrasaction: Literal['raise', 'ignore'] = 'ignore', sort_keys: bool = False, fail_if_empty: bool = False, attempts: int = 3, delay: float = 0.35, backoff: float = 1.5, max_delay: float | None = 2.0, **kwargs) → str | PathLike | Path | None¶

Retry write_csv_atomic() after transient failures.

Parameters

Name	Type	Description
data	List[Dict[str, Any]]	Row mappings to write.
delimiter	str	Field separator.
encoding	str	Target text encoding.
include_header	bool	Whether to write headers.
headers	List[str] \| None	Explicit field order.
quoting	int	`csv` quoting mode.
lineterminator	str	Row terminator.
extrasaction	Literal['raise', 'ignore']	Handling for extra row keys.
sort_keys	bool	Whether to sort headers.
fail_if_empty	bool	Whether empty data should cancel the write.
attempts	int	Total attempts, including the first.
delay	float	Initial retry delay in seconds.
backoff	float	Delay multiplier after each failure.
max_delay	float \| None	Optional upper delay bound.
**kwargs		Additional `csv.DictWriter` arguments.

Returns

Type	Description
str \| PathLike \| Path \| None	The target path after success; otherwise `None`.

Examples

Retry atomic publication of a CSV dataset:

path = FileManager("results.csv").write_csv_atomic_retry(
    [{"id": 1}], attempts=5
)

watch(callback: Callable[[FileManager], None], *, interval: float = 1.0, stop_event: Any | None = None, run_in_thread: bool = True) → Any¶

Call a function whenever the target modification time changes.

The implementation uses watchdog when installed and otherwise polls the target. Callback exceptions are intentionally suppressed.

Parameters

Name	Type	Description
callback	Callable[[FileManager], None]	Callable receiving this manager after a detected change.
interval	float	Polling interval when `watchdog` is unavailable.
stop_event	Any \| None	Event used to stop monitoring. A new event is created when omitted.
run_in_thread	bool	Whether to run the monitoring loop in a daemon thread.

Returns

Type	Description
Any	The started thread in background mode; otherwise `None` after the blocking monitoring loop exits.

Examples

Monitor until an event is set:

stop = threading.Event()
thread = file.watch(on_change, stop_event=stop)
stop.set()

stream_jsonl(encoding: str = 'utf-8', skip_errors: bool = True) → Iterator[Any]¶

Yield decoded non-empty JSON Lines records.

Parameters

Name	Type	Description
encoding	str	Source text encoding.
skip_errors	bool	Whether malformed records should be ignored.

Yields

The decoded Python value for each valid non-empty line.

Raises

Exception	Description
json.JSONDecodeError	A malformed record is encountered while `skip_errors` is `False`.
OSError	The source cannot be opened or read.

Examples

Process records without loading the file into memory:

for record in FileManager("results.jsonl").stream_jsonl():
    process(record)

checksum(algorithm: str = 'sha256', chunk_size: int = 65536, *, usedforsecurity: bool = True) → str¶

Compute a hexadecimal file digest by streaming fixed-size chunks.

Parameters

Name	Type	Description
algorithm	str	Hash algorithm accepted by the package hasher policy.
chunk_size	int	Number of bytes read per iteration.
usedforsecurity	bool	Whether security policy should reject weak hashes such as MD5 and SHA-1.

Returns

Type	Description
str	Lowercase hexadecimal digest.

Examples

Compute SHA-256, or explicitly permit a legacy checksum:

secure_digest = file.checksum()
legacy_digest = file.checksum("md5", usedforsecurity=False)

compress(dst: str | PathLike | Path | None = None, *, level: int = 9) → FileManager¶

Stream the target into a gzip file.

Parameters

Name	Type	Description
dst	str \| PathLike \| Path \| None	Output path. Defaults to the source path plus `.gz`.
level	int	gzip compression level.

Returns

Type	Description
FileManager	A manager for the compressed output.

Raises

Exception	Description
OSError	The source cannot be read or the output cannot be written.

Examples

Create results.json.gz:

archive = FileManager("results.json").compress()

decompress(dst: str | PathLike | Path | None = None) → FileManager¶

Stream a gzip target into an uncompressed file.

Parameters

Name	Type	Description
dst	str \| PathLike \| Path \| None	Output path. By default a trailing `.gz` suffix is removed.

Returns

Type	Description
FileManager	A manager for the uncompressed output.

Raises

Exception	Description
OSError	The source cannot be read or the output cannot be written.

Examples

Restore results.json from results.json.gz:

restored = FileManager("results.json.gz").decompress()