A safer writer for files and streams that prevents corruption and partial writes through atomic operations
npx @tessl/cli install tessl/pypi-safer@5.1.0A safer writer for files and streams that prevents corruption and partial writes through atomic operations. Safer provides drop-in replacements for Python's built-in file operations, ensuring that either complete data is written or the original file remains unchanged.
pip install saferimport saferAll functions are available directly from the safer module:
from safer import open, writer, closer, dump, printerimport safer
import json
# Safe file writing - either complete or nothing
with safer.open('data.json', 'w') as fp:
json.dump({"key": "value"}, fp)
# If an exception occurs here, data.json remains unchanged
# Safe stream wrapping
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
with safer.writer(sock) as s:
s.write(b"header\n")
s.write(b"body\n") # If exception occurs, nothing is sent to socket
s.write(b"footer\n")
# Safe printing to file
with safer.printer('output.txt', 'w') as print_func:
for item in data:
print_func(item) # Either all items printed or file unchangedDrop-in replacement for Python's built-in open() function that provides atomic file writing operations.
def open(
name: Path | str,
mode: str = 'r',
buffering: int = -1,
encoding: str | None = None,
errors: str | None = None,
newline: str | None = None,
closefd: bool = True,
opener: t.Callable | None = None,
make_parents: bool = False,
delete_failures: bool = True,
temp_file: bool = False,
dry_run: bool | t.Callable = False,
enabled: bool = True,
) -> t.IO:
"""
Safe file opening with atomic write operations.
Parameters:
- name: File path to open
- mode: File open mode (same as built-in open)
- buffering: Buffer size (-1 for default)
- encoding: Text encoding for text mode
- errors: Error handling strategy
- newline: Newline handling
- closefd: Whether to close file descriptor
- opener: Custom opener function
- make_parents: Create parent directories if needed
- delete_failures: Delete temporary files on failure
- temp_file: Use disk temporary file instead of memory buffer
- dry_run: Test mode - don't actually write (bool) or pass data to callable
- enabled: Enable/disable safer functionality
Returns:
File-like object that writes atomically
"""Wraps existing streams, sockets, or callables to provide safe writing operations that only commit on successful completion.
def writer(
stream: t.Callable | None | t.IO | Path | str = None,
is_binary: bool | None = None,
close_on_exit: bool = False,
temp_file: bool = False,
chunk_size: int = 0x100000,
delete_failures: bool = True,
dry_run: bool | t.Callable = False,
enabled: bool = True,
) -> t.Callable | t.IO:
"""
Write safely to file streams, sockets and callables.
Parameters:
- stream: Target stream, socket, callable, or file path (None for sys.stdout)
- is_binary: Whether stream is binary (auto-detected if None)
- close_on_exit: Close underlying stream when writer closes
- temp_file: Use disk temporary file (bool or custom path)
- chunk_size: Chunk size for temporary file transfers
- delete_failures: Delete temporary files on failure
- dry_run: Test mode - don't write (bool) or pass data to callable
- enabled: Enable/disable safer functionality
Returns:
Wrapped stream that writes atomically
"""Like writer() but with automatic closing of the underlying stream enabled by default.
def closer(
stream: t.IO,
is_binary: bool | None = None,
close_on_exit: bool = True,
**kwds
) -> t.Callable | t.IO:
"""
Safe stream wrapper with automatic closing.
Parameters:
- stream: Stream to wrap
- is_binary: Whether stream is binary (auto-detected if None)
- close_on_exit: Close underlying stream when done (default True)
- **kwds: Additional arguments passed to writer()
Returns:
Wrapped stream with automatic closing
"""Safe serialization using json.dump or other serialization protocols with atomic write operations.
def dump(
obj,
stream: t.Callable | None | t.IO | Path | str = None,
dump: t.Any = None,
**kwargs,
) -> t.Any:
"""
Safely serialize objects to streams or files.
Parameters:
- obj: Object to serialize
- stream: Target stream, file path, or None for sys.stdout
- dump: Serialization function/module (defaults to json.dump)
Can be 'json', 'yaml', 'toml' or callable
- **kwargs: Additional arguments passed to dump function
Returns:
Result from the dump function
"""Context manager that yields a print function for safe file writing, ensuring atomic operations.
def printer(
name: Path | str,
mode: str = 'w',
*args,
**kwargs
) -> t.Iterator[t.Callable]:
"""
Context manager for safe printing to files.
Parameters:
- name: File path
- mode: File open mode (must support writing)
- *args, **kwargs: Additional arguments passed to safer.open()
Yields:
Print function that writes to the opened file
Usage:
with safer.printer('output.txt', 'w') as print_func:
print_func("Line 1")
print_func("Line 2")
"""from pathlib import Path
from typing import IO, Callable, Any, Iterator
# Common type aliases used throughout the API
StreamType = Callable | None | IO | Path | str
DumpFunction = str | Callable | Any
FileMode = str # e.g., 'r', 'w', 'a', 'rb', 'wb', 'r+', etc.
PrintFunction = Callable[..., None] # Function returned by printer()