Safe File Operations¶

A shared service for atomic, crash-safe file operations integrated with the path management system.

Table of Contents¶

Quick Start
Core Operations
JSON Operations
MessagePack Operations
Text Operations
Exception Handling
Thread Safety
Windows Compatibility

Quick Start¶

from src.shared_services.file_operations.api import (
    save_json, load_json,
    save_msgpack, load_msgpack,
    save_text, load_text,
)
from src.shared_services.constants.paths import MyConfigPath  # PathDef

# Save JSON atomically
config = {"key": "value", "german": "Groesse"}
save_json(MyConfigPath, config)

# Load JSON
loaded = load_json(MyConfigPath)
print(loaded["key"])  # "value"

Core Operations¶

For raw bytes operations when you handle serialization yourself.

atomic_write¶

from src.shared_services.file_operations.api import atomic_write

# Write bytes atomically
atomic_write(DataPath, b"binary data here")

# Without fsync (faster but less safe)
atomic_write(DataPath, data, fsync=False)

atomic_read¶

from src.shared_services.file_operations.api import atomic_read

data = atomic_read(DataPath)

JSON Operations¶

Pass any JSON-serializable object (dict, list, etc.).

save_json¶

from src.shared_services.file_operations.api import save_json

# Save dictionary
config = {"key": "value", "items": [1, 2, 3]}
save_json(ConfigPath, config)

# Custom indent
save_json(ConfigPath, config, indent=4)

load_json¶

from src.shared_services.file_operations.api import load_json

config = load_json(ConfigPath)
print(config["key"])

MessagePack Operations¶

For efficient binary serialization of Python objects.

save_msgpack¶

from src.shared_services.file_operations.api import save_msgpack

data = {"items": [1, 2, 3], "nested": {"key": "value"}}
save_msgpack(DataPath, data)

load_msgpack¶

from src.shared_services.file_operations.api import load_msgpack

data = load_msgpack(DataPath)

Text Operations¶

For plain text files.

save_text¶

from src.shared_services.file_operations.api import save_text

save_text(LogPath, "Application started\n")

# Custom encoding
save_text(LogPath, text, encoding="latin-1")

load_text¶

from src.shared_services.file_operations.api import load_text

content = load_text(LogPath)

Exception Handling¶

All exceptions inherit from FileOperationError.

Exception Hierarchy¶

Exception	Description
`FileOperationError`	Base exception for all file operations
`AtomicWriteError`	Write operation failures
`FileLockError`	File locked by another process
`SerializationError`	JSON/msgpack serialization errors

Basic Error Handling¶

from src.shared_services.file_operations.api import (
    save_json,
    FileOperationError,
)

try:
    save_json(ConfigPath, config)
except FileOperationError as e:
    print(f"File operation failed: {e}")

Specific Error Handling¶

from src.shared_services.file_operations.api import (
    save_json,
    load_json,
    AtomicWriteError,
    FileLockError,
    SerializationError,
)

try:
    save_json(ConfigPath, config)
except SerializationError:
    print("Object cannot be serialized to JSON")
except FileLockError:
    print("File is locked by another process")
except AtomicWriteError:
    print("Write operation failed")

try:
    data = load_json(DataPath)
except FileNotFoundError:
    print("File does not exist")
except SerializationError:
    print("Invalid JSON format")

Thread Safety¶

All operations are thread-safe. Each file path has its own lock, allowing concurrent writes to different files while preventing conflicts on the same file.

import threading
from src.shared_services.file_operations.api import save_json

def worker(path_def, data):
    save_json(path_def, data)  # Thread-safe

# Multiple threads can safely write to different files
threads = [
    threading.Thread(target=worker, args=(Path1, data1)),
    threading.Thread(target=worker, args=(Path2, data2)),
]
for t in threads:
    t.start()

Windows Compatibility¶

The atomic write pattern handles Windows-specific issues:

Antivirus scanning: Windows Defender may briefly lock new files
File indexing: Windows Search may hold file handles
Solution: Exponential backoff retry (0.1s, 0.2s, 0.4s, 0.8s, 1.6s)

If a file cannot be replaced after 5 retries, FileLockError is raised.

Type Safety¶

All functions only accept PathDef objects - raw strings are rejected:

from src.shared_services.file_operations.api import save_json

# Correct - use PathDef from constants
save_json(ConfigPath, {"key": "value"})

# Wrong - raw string raises TypeError
save_json("config.json", {"key": "value"})  # TypeError!

Full Import Reference¶

from src.shared_services.file_operations.api import (
    # Core operations
    atomic_write,
    atomic_read,
    # JSON
    save_json,
    load_json,
    # MessagePack
    save_msgpack,
    load_msgpack,
    # Text
    save_text,
    load_text,
    # Exceptions
    FileOperationError,
    AtomicWriteError,
    FileLockError,
    SerializationError,
)

API Reference¶

`src.shared_services.file_operations.api` ¶

Public API for safe file operations.

USAGE

from src.shared_services.file_operations.api import ( save_json, load_json, save_msgpack, load_msgpack, save_text, load_text, atomic_write, atomic_read, FileOperationError, ) from src.shared_services.constants.paths import MyConfigPath # PathDef

Save JSON atomically¶

config = {"key": "value", "german": "Größe"} save_json(MyConfigPath, config)

Load JSON¶

loaded = load_json(MyConfigPath)

Save/load msgpack¶

save_msgpack(DataPath, {"items": [1, 2, 3]}) data = load_msgpack(DataPath)

Save/load text¶

save_text(TextPath, "Hello World") text = load_text(TextPath)

NOTE

All functions ONLY accept PathDef objects from constants/paths.py. Raw strings are not supported - this is by design for type safety.

`AtomicWriteError` ¶

Bases: FileOperationError

Raised when an atomic write operation fails.

Source code in src\shared_services\file_operations\exceptions.py

class AtomicWriteError(FileOperationError):
    """Raised when an atomic write operation fails."""

    pass

`FileLockError` ¶

Bases: FileOperationError

Raised when a file is locked by another process.

Source code in src\shared_services\file_operations\exceptions.py

class FileLockError(FileOperationError):
    """Raised when a file is locked by another process."""

    pass

`FileOperationError` ¶

Bases: Exception

Base exception for all file operations.

Source code in src\shared_services\file_operations\exceptions.py

class FileOperationError(Exception):
    """Base exception for all file operations."""

    pass

`SerializationError` ¶

Bases: FileOperationError

Raised when serialization or deserialization fails.

Source code in src\shared_services\file_operations\exceptions.py

class SerializationError(FileOperationError):
    """Raised when serialization or deserialization fails."""

    pass

`atomic_read(path_def)` ¶

Read file contents as bytes.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the file to read.	required

Returns:

Type	Description
`bytes`	File contents as bytes.

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`FileNotFoundError`	If file does not exist.

Example

data = atomic_read(MyDataPath)

Source code in src\shared_services\file_operations\api.py

def atomic_read(path_def: "PathDef") -> bytes:
    """
    Read file contents as bytes.

    Args:
        path_def: PathDef for the file to read.

    Returns:
        File contents as bytes.

    Raises:
        TypeError: If path_def is not a PathDef.
        FileNotFoundError: If file does not exist.

    Example:
        data = atomic_read(MyDataPath)
    """
    path = _validate_path_def(path_def, "atomic_read")
    return atomic_read_internal(path)

`atomic_write(path_def, data, *, fsync=True)` ¶

Write bytes atomically to a file.

Uses safe atomic write pattern: temp file, fsync, atomic replace. Thread-safe with per-file locking.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the target file.	required
`data`	`bytes`	Bytes to write.	required
`fsync`	`bool`	Sync to disk before replace (default True).	`True`

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`AtomicWriteError`	If the write fails.
`FileLockError`	If file is locked.

Example

from src.shared_services.constants.paths import MyDataPath atomic_write(MyDataPath, b "binary data here")

Source code in src\shared_services\file_operations\api.py

def atomic_write(
    path_def: "PathDef",
    data: bytes,
    *,
    fsync: bool = True,
) -> None:
    """
    Write bytes atomically to a file.

    Uses safe atomic write pattern: temp file, fsync, atomic replace.
    Thread-safe with per-file locking.

    Args:
        path_def: PathDef for the target file.
        data: Bytes to write.
        fsync: Sync to disk before replace (default True).

    Raises:
        TypeError: If path_def is not a PathDef.
        AtomicWriteError: If the write fails.
        FileLockError: If file is locked.

    Example:
        from src.shared_services.constants.paths import MyDataPath
        atomic_write(MyDataPath, b "binary data here")
    """
    path = _validate_path_def(path_def, "atomic_write")
    atomic_write_internal(path, data, fsync=fsync)

`load_json(path_def)` ¶

Load JSON file and return object.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the JSON file.	required

Returns:

Type	Description
`Any`	Deserialized object (dict, list, etc.).

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`FileNotFoundError`	If file does not exist.
`SerializationError`	If JSON is invalid.

Example

config = load_json(ConfigPath) print(config["key"])

Source code in src\shared_services\file_operations\api.py

def load_json(path_def: "PathDef") -> Any:
    """
    Load JSON file and return object.

    Args:
        path_def: PathDef for the JSON file.

    Returns:
        Deserialized object (dict, list, etc.).

    Raises:
        TypeError: If path_def is not a PathDef.
        FileNotFoundError: If file does not exist.
        SerializationError: If JSON is invalid.

    Example:
        config = load_json(ConfigPath)
        print(config["key"])
    """
    path = _validate_path_def(path_def, "load_json")

    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except json.JSONDecodeError as e:
        raise SerializationError(f"Invalid JSON in {path}: {e}") from e

`load_msgpack(path_def)` ¶

Load MessagePack file and return object.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the msgpack file.	required

Returns:

Type	Description
`Any`	Deserialized object.

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`FileNotFoundError`	If file does not exist.
`SerializationError`	If msgpack is invalid.

Example

data = load_msgpack(DataPath) print(data["items"])

Source code in src\shared_services\file_operations\api.py

def load_msgpack(path_def: "PathDef") -> Any:
    """
    Load MessagePack file and return object.

    Args:
        path_def: PathDef for the msgpack file.

    Returns:
        Deserialized object.

    Raises:
        TypeError: If path_def is not a PathDef.
        FileNotFoundError: If file does not exist.
        SerializationError: If msgpack is invalid.

    Example:
        data = load_msgpack(DataPath)
        print(data["items"])
    """
    path = _validate_path_def(path_def, "load_msgpack")

    try:
        with open(path, "rb") as f:
            return msgpack.unpackb(f.read(), raw=False)
    except msgpack.UnpackException as e:
        raise SerializationError(f"Invalid msgpack in {path}: {e}") from e

`load_text(path_def, *, encoding='utf-8')` ¶

Load text file content.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the text file.	required
`encoding`	`str`	Text encoding (default utf-8).	`'utf-8'`

Returns:

Type	Description
`str`	File content as string.

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`FileNotFoundError`	If file does not exist.

Example

content = load_text(LogPath)

Source code in src\shared_services\file_operations\api.py

def load_text(
    path_def: "PathDef",
    *,
    encoding: str = "utf-8",
) -> str:
    """
    Load text file content.

    Args:
        path_def: PathDef for the text file.
        encoding: Text encoding (default utf-8).

    Returns:
        File content as string.

    Raises:
        TypeError: If path_def is not a PathDef.
        FileNotFoundError: If file does not exist.

    Example:
        content = load_text(LogPath)
    """
    path = _validate_path_def(path_def, "load_text")
    return path.read_text(encoding=encoding)

`save_json(path_def, obj, *, indent=2)` ¶

Save object as JSON atomically.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the target file.	required
`obj`	`Any`	Object to serialize (dict, list, etc.).	required
`indent`	`int`	JSON indentation (default 2).	`2`

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`SerializationError`	If object cannot be serialized.
`AtomicWriteError`	If write fails.

Example

config = {"key": "value", "items": [1, 2, 3]} save_json(ConfigPath, config)

Source code in src\shared_services\file_operations\api.py

def save_json(
    path_def: "PathDef",
    obj: Any,
    *,
    indent: int = 2,
) -> None:
    """
    Save object as JSON atomically.

    Args:
        path_def: PathDef for the target file.
        obj: Object to serialize (dict, list, etc.).
        indent: JSON indentation (default 2).

    Raises:
        TypeError: If path_def is not a PathDef.
        SerializationError: If object cannot be serialized.
        AtomicWriteError: If write fails.

    Example:
        config = {"key": "value", "items": [1, 2, 3]}
        save_json(ConfigPath, config)
    """
    path = _validate_path_def(path_def, "save_json")

    try:
        data = json.dumps(obj, indent=indent, ensure_ascii=False).encode("utf-8")
    except (TypeError, ValueError) as e:
        raise SerializationError(f"Failed to serialize JSON: {e}") from e

    atomic_write_internal(path, data)

`save_msgpack(path_def, obj)` ¶

Save object as MessagePack atomically.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the target file.	required
`obj`	`Any`	Object to serialize.	required

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`SerializationError`	If object cannot be serialized.
`AtomicWriteError`	If write fails.

Example

data = {"items": [1, 2, 3], "nested": {"key": "value"}} save_msgpack(DataPath, data)

Source code in src\shared_services\file_operations\api.py

def save_msgpack(path_def: "PathDef", obj: Any) -> None:
    """
    Save object as MessagePack atomically.

    Args:
        path_def: PathDef for the target file.
        obj: Object to serialize.

    Raises:
        TypeError: If path_def is not a PathDef.
        SerializationError: If object cannot be serialized.
        AtomicWriteError: If write fails.

    Example:
        data = {"items": [1, 2, 3], "nested": {"key": "value"}}
        save_msgpack(DataPath, data)
    """
    path = _validate_path_def(path_def, "save_msgpack")

    try:
        data = msgpack.packb(obj, use_bin_type=True)
    except (TypeError, ValueError) as e:
        raise SerializationError(f"Failed to serialize msgpack: {e}") from e

    atomic_write_internal(path, data)

`save_text(path_def, text, *, encoding='utf-8')` ¶

Save text string atomically.

Parameters:

Name	Type	Description	Default
`path_def`	`PathDef`	PathDef for the target file.	required
`text`	`str`	Text content to write.	required
`encoding`	`str`	Text encoding (default utf-8).	`'utf-8'`

Raises:

Type	Description
`TypeError`	If path_def is not a PathDef.
`AtomicWriteError`	If write fails.

Example

save_text(LogPath, "Application started\n")

Source code in src\shared_services\file_operations\api.py

def save_text(
    path_def: "PathDef",
    text: str,
    *,
    encoding: str = "utf-8",
) -> None:
    """
    Save text string atomically.

    Args:
        path_def: PathDef for the target file.
        text: Text content to write.
        encoding: Text encoding (default utf-8).

    Raises:
        TypeError: If path_def is not a PathDef.
        AtomicWriteError: If write fails.

    Example:
        save_text(LogPath, "Application started\\n")
    """
    path = _validate_path_def(path_def, "save_text")
    data = text.encode(encoding)
    atomic_write_internal(path, data)

Safe File Operations¶

Table of Contents¶

Quick Start¶

Core Operations¶

atomic_write¶

atomic_read¶

JSON Operations¶

save_json¶

load_json¶

MessagePack Operations¶

save_msgpack¶

load_msgpack¶

Text Operations¶

save_text¶

load_text¶

Exception Handling¶

Exception Hierarchy¶

Basic Error Handling¶

Specific Error Handling¶

Thread Safety¶

Windows Compatibility¶

Type Safety¶

Full Import Reference¶

API Reference¶

src.shared_services.file_operations.api ¶

Save JSON atomically¶

Load JSON¶

Save/load msgpack¶

Save/load text¶

AtomicWriteError ¶

FileLockError ¶

FileOperationError ¶

SerializationError ¶

atomic_read(path_def) ¶

atomic_write(path_def, data, *, fsync=True) ¶

load_json(path_def) ¶

load_msgpack(path_def) ¶

load_text(path_def, *, encoding='utf-8') ¶

save_json(path_def, obj, *, indent=2) ¶

save_msgpack(path_def, obj) ¶

save_text(path_def, text, *, encoding='utf-8') ¶

`src.shared_services.file_operations.api` ¶

`AtomicWriteError` ¶

`FileLockError` ¶

`FileOperationError` ¶

`SerializationError` ¶

`atomic_read(path_def)` ¶

`atomic_write(path_def, data, *, fsync=True)` ¶

`load_json(path_def)` ¶

`load_msgpack(path_def)` ¶

`load_text(path_def, *, encoding='utf-8')` ¶

`save_json(path_def, obj, *, indent=2)` ¶

`save_msgpack(path_def, obj)` ¶

`save_text(path_def, text, *, encoding='utf-8')` ¶