atomicio 1.0.1
Thread-safe, async-safe, and atomic file operations with extensible format support for Python
License: MIT
Requires Python: >=3.8
Keywords: atomic, file-operations, thread-safe, async, locking, concurrent, file-locking, safe-write, data-integrity

# 🗃️ Atomicio

[![Python Support](https://img.shields.io/pypi/pyversions/atomicio.svg)](https://pypi.org/project/atomicio/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Code Style: Black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)

> 🎯 **Thread-safe, async-safe, and atomic file operations for Python - because data integrity should never be compromised!**

**Atomicio** is a comprehensive Python package that provides robust, thread-safe, and async-safe file operations with proper locking mechanisms. Whether you're building a simple script or a complex multi-threaded application, Atomicio ensures your file operations are reliable, consistent, and corruption-free.

**🔌 Extensible Plugin System**: Seamlessly add support for any file format (CSV, XML, binary, custom) with simple loader/dumper functions. All safety guarantees apply to custom formats automatically!

---

## 🤔 Which Class Should I Use?

<details>
<summary><strong>📖 Click here for a quick decision guide with real-world examples</strong></summary>

### 🎯 **Quick Decision Tree**

**Ask yourself these questions:**

1. **Are you using `async`/`await` in your code?**
   - ✅ **YES** → Use **`AsyncSafeFile`**
   - ❌ **NO** → Continue to question 2

2. **Do you need to perform multiple operations while holding the lock?**
   - ✅ **YES** → Use **`ThreadedSafeFile`**
   - ❌ **NO** → Use **`SafeFile`**

---

### 🔒 **Use `SafeFile` when:**

**Perfect for:** Simple scripts, config management, basic data persistence

**Real-world scenarios:**
- ✅ Saving user preferences in a desktop app
- ✅ Storing API tokens or configuration files
- ✅ Quick scripts that read/write JSON, YAML, or TOML
- ✅ Single-operation file access with minimal complexity

**Example: Configuration Manager**
```python
from atomicio import SafeFile

# Simple config management - each operation is independent
config = SafeFile('app_config.json')

# Read current settings
settings = config.read() or {}

# Update a setting (single atomic operation)
settings['theme'] = 'dark'
config.write(settings)

# Another independent update
settings = config.read() or {}
settings['language'] = 'en'
config.write(settings)
```

**Why SafeFile here?** Each `read()` and `write()` is independent - no need to hold locks across operations.

---

### 🔗 **Use `ThreadedSafeFile` when:**

**Perfect for:** Data processing pipelines, complex workflows, multi-step operations

**Real-world scenarios:**
- ✅ Processing data in stages where intermediate steps must be atomic
- ✅ Read → Modify → Verify → Write workflows
- ✅ Multi-threaded applications with complex file coordination
- ✅ Operations where you need to "lock, do multiple things, then unlock"

**Example: Data Processing Pipeline**
```python
from atomicio import ThreadedSafeFile

# Process data in multiple steps while holding the lock
data_file = ThreadedSafeFile('pipeline_data.json', timeout=10.0)

with data_file.locked() as f:
    # Lock is held during this entire block

    # Step 1: Read raw data
    data = f.read() or {}

    # Step 2: Validate
    if not validate_data(data):
        raise ValueError("Invalid data")

    # Step 3: Transform
    data['processed'] = transform(data)
    f.write(data)  # Save intermediate result

    # Step 4: Enrich
    data['enriched'] = enrich(data)
    f.write(data)  # Final save

    # Lock automatically released here
```

**Why ThreadedSafeFile here?** Multiple operations (read, validate, transform, write multiple times) must happen atomically as a sequence. No other thread can access the file during this entire workflow.

**❌ Wrong approach with SafeFile:**
```python
# DON'T DO THIS - Race condition risk!
data_file = SafeFile('pipeline_data.json')

data = data_file.read()  # Lock released here
# ⚠️ Another thread could modify the file here!
data['processed'] = transform(data)
data_file.write(data)  # Lock acquired again - but data might be stale!
```

---

### ⚡ **Use `AsyncSafeFile` when:**

**Perfect for:** Web servers, async APIs, event-driven applications, high-concurrency services

**Real-world scenarios:**
- ✅ FastAPI / Quart / Sanic web applications
- ✅ Discord bots, Telegram bots (async frameworks)
- ✅ WebSocket servers handling multiple connections
- ✅ Any application using `asyncio` event loops
- ✅ High-concurrency services with async I/O

**Example: FastAPI Endpoint**
```python
from fastapi import FastAPI
from atomicio import AsyncSafeFile

app = FastAPI()
analytics = AsyncSafeFile('user_analytics.json', timeout=5.0)

@app.post("/track-event/")
async def track_event(user_id: str, event: str):
    # Non-blocking async file operations
    async with analytics.locked() as f:
        # Lock held during entire block (async)

        # Read current analytics
        data = await f.read() or {}

        # Update user events
        if user_id not in data:
            data[user_id] = []
        data[user_id].append({
            'event': event,
            'timestamp': time.time()
        })

        # Save updated analytics
        await f.write(data)

    return {"status": "tracked"}

# FastAPI can handle hundreds of concurrent requests
# without blocking thanks to AsyncSafeFile
```

**Why AsyncSafeFile here?** FastAPI is async - using `SafeFile` or `ThreadedSafeFile` would block the event loop and kill performance. `AsyncSafeFile` allows hundreds of concurrent requests without blocking.

**❌ Wrong approach with ThreadedSafeFile:**
```python
# DON'T DO THIS in async code!
@app.post("/track-event/")
async def track_event(user_id: str, event: str):
    analytics = ThreadedSafeFile('user_analytics.json')

    # ⚠️ This blocks the entire event loop!
    with analytics.locked() as f:
        data = f.read()  # Blocking I/O - terrible for async!
        # While this request waits, ALL other requests are blocked!
```

---

### 📊 **Quick Comparison Table**

| Scenario | SafeFile | ThreadedSafeFile | AsyncSafeFile |
|----------|----------|------------------|---------------|
| **Simple config read/write** | ✅ Perfect | ⚠️ Overkill | ❌ Wrong tool |
| **Multi-step data processing** | ❌ Race conditions | ✅ Perfect | ❌ Wrong tool |
| **FastAPI/async web server** | ❌ Blocks event loop | ❌ Blocks event loop | ✅ Perfect |
| **Desktop app settings** | ✅ Perfect | ⚠️ Overkill | ❌ Wrong tool |
| **Multi-threaded pipeline** | ❌ No cross-op locking | ✅ Perfect | ❌ Wrong tool |
| **Discord/Telegram bot** | ❌ Blocks event loop | ❌ Blocks event loop | ✅ Perfect |
| **Simple script** | ✅ Perfect | ⚠️ Overkill | ❌ Unnecessary complexity |
| **High-concurrency async** | ❌ Poor performance | ❌ Poor performance | ✅ Perfect |

**Legend:**
- ✅ **Perfect**: Best choice for this scenario
- ⚠️ **Overkill**: Works but unnecessarily complex
- ❌ **Wrong tool**: Don't use - will cause issues

---

### 💡 **Still Not Sure?**

**Default recommendations:**
1. **Writing a regular Python script or app?** → Start with **`SafeFile`**
2. **Using `async`/`await` anywhere?** → Use **`AsyncSafeFile`**
3. **Need complex multi-step operations?** → Use **`ThreadedSafeFile`**

**When in doubt:** Start with `SafeFile` - it's the simplest. Upgrade to `ThreadedSafeFile` only when you need cross-operation locking. Use `AsyncSafeFile` only in async contexts.

</details>

---

## ✨ Overview & Class Comparison

### 🔐 **Three Specialized Classes for Different Needs**

Atomicio provides three distinct classes, each optimized for specific use cases and concurrency requirements:

| Class | **SafeFile** | **ThreadedSafeFile** | **AsyncSafeFile** |
|-------|-------------|---------------------|-------------------|
| **🎯 Primary Use** | Basic atomic operations | Complex cross-operation locking | Async applications |
| **🔧 Locking Strategy** | FileLock (inter-process) + Global thread lock | Per-file RLock + FileLock | Per-file AsyncLock + FileLock |
| **⚡ Concurrency** | Thread-safe individual operations | Thread-safe operation sequences | Async-safe operation sequences |
| **🔄 Context Manager** | Basic context (`with SafeFile()`) | Advanced locked context (`with tsf.locked()`) | Async locked context (`async with asf.locked()`) |
| **📊 Performance** | Fastest for simple operations | Optimized for complex workflows | Best for async environments |
| **🎪 Complexity** | Simple and lightweight | Advanced coordination | Async-first design |

### 🧠 **Core Technologies Behind Each Class**

#### **🔒 SafeFile** - Simple Atomic Operations
- **FileLock**: Provides inter-process safety using file system locks
- **Global Thread Lock**: Ensures thread safety within the same process
- **Atomic Write**: Uses temporary files with atomic rename operations
- **Format Detection**: Automatic serialization based on file extensions

#### **🔗 ThreadedSafeFile** - Cross-Operation Coordination
- **Per-File RLock**: Individual threading.RLock for each file path
- **Reentrant Locking**: Same thread can acquire the same lock multiple times
- **Lock Persistence**: Holds locks across multiple operations in `locked()` context
- **Timeout Control**: Configurable timeout for lock acquisition

#### **⚡ AsyncSafeFile** - Async-Native Operations
- **Per-File AsyncLock**: Individual asyncio.Lock for each file path
- **Async Context Managers**: Native `async with` support
- **Non-Blocking Operations**: Designed for asyncio event loops
- **Async Timeout Control**: Proper async timeout handling with cancellation

## 🚀 Quick Start

### Installation

```bash
pip install atomicio
```

### 🎯 Quick Examples

```python
from atomicio import SafeFile, ThreadedSafeFile, AsyncSafeFile
from atomicio import register_format, list_supported_formats  # Plugin system

# 🔒 SafeFile - Simple atomic operations
sf = SafeFile('config.json')
sf.write({'setting': 'value'})
data = sf.read()

# 🔗 ThreadedSafeFile - Complex coordinated operations
tsf = ThreadedSafeFile('data.yaml', timeout=5.0)
with tsf.locked() as f:
    data = f.read() or {}
    data['processed'] = True
    f.write(data)
    # Lock held during entire operation sequence

# ⚡ AsyncSafeFile - Async operations
import asyncio

async def async_example():
    asf = AsyncSafeFile('async_data.json')
    async with asf.locked() as f:
        data = await f.read() or {}
        data['timestamp'] = time.time()
        await f.write(data)

asyncio.run(async_example())

# 🔌 Plugin System - Extend to any file format
def csv_loader(f): return list(csv.DictReader(f))
def csv_dumper(data, f): csv.DictWriter(f, data[0].keys()).writerows(data)

register_format('.csv', csv_loader, csv_dumper)
# Now CSV files work with all classes: SafeFile('data.csv'), etc.
```

---

## 🔒 SafeFile - Simple Atomic Operations

### 🧠 Working Principle

SafeFile is designed for applications that need **basic atomic file operations** with minimal complexity. It combines FileLock for inter-process safety with a global thread lock for intra-process coordination.

**Key Concepts:**
- **Atomic Writes**: Every write operation uses a temporary file that's atomically renamed
- **Format Auto-Detection**: Automatically serializes/deserializes based on file extension (.json, .yaml, .toml, .txt)
- **Inter-Process Safety**: FileLock prevents conflicts between different processes
- **Thread Safety**: Global lock ensures thread safety within the same process
- **Simplicity First**: Minimal API surface for straightforward use cases

**When to Use SafeFile:**
- ✅ Simple configuration file management
- ✅ Basic data persistence needs
- ✅ Applications with minimal concurrency requirements
- ✅ Quick prototyping and scripting
- ❌ Complex multi-step operations requiring lock persistence
- ❌ High-concurrency applications with complex coordination needs

### 📋 Complete API Reference

#### **Constructor**
```python
SafeFile(path, timeout=True)
```
- `path`: File path (str or Path object)
- `timeout`: Lock acquisition timeout behavior:
  - `True` (default): 15 seconds timeout
  - `False` or `None`: Wait forever (infinite blocking)
  - `int` or `float`: Wait that many seconds (e.g., `5.0`)

**Examples:**
```python
SafeFile('config.json')              # 15 second timeout (default)
SafeFile('config.json', timeout=5.0) # 5 second timeout
SafeFile('config.json', timeout=None) # Wait forever
```

#### **Core Methods**
```python
# Read/Write Operations
read() -> Any | None                    # Read and deserialize file content
write(data: Any) -> None               # Write and serialize data atomically
update(data: Any) -> None              # Update existing data (merge for dicts)

# Binary Operations
read_bytes() -> bytes | None           # Read raw bytes
write_bytes(data: bytes) -> None       # Write raw bytes

# Text Operations
append(text: str) -> None              # Append text to file

# Utility Methods
@staticmethod
list_supported_formats() -> list      # List supported file formats ['.json', '.yaml', ...]
```

#### **Context Manager Support**
```python
# Basic context manager
with SafeFile('config.json') as sf:
    data = sf.read()
    sf.write(updated_data)
```

#### **Supported File Formats**
- **`.json`** - JSON format with automatic pretty-printing
- **`.yaml/.yml`** - YAML format with safe loading
- **`.toml`** - TOML format support
- **`.txt`** - Plain text format
- **🔌 Custom formats** - Add any format via the plugin system (CSV, XML, binary, etc.)

#### **Custom Exceptions**
SafeFile raises specific exceptions for different error conditions:
- **`FileReadError`** - When file reading fails (corrupt file, permission issues, etc.)
- **`FileWriteError`** - When file writing fails (disk full, permission denied, etc.)
- **`FileAppendError`** - When appending to file fails (similar to write errors)

### 🎯 Use Case Example: Configuration Manager

<details>
<summary><strong>📖 Click to expand: Complete Configuration Manager Example</strong></summary>

**Scenario**: A web application needs to manage user preferences that can be updated by multiple processes (web workers) and should persist across restarts.

**Why SafeFile is Perfect:**
- Simple key-value storage requirements
- Multiple processes need safe access
- Atomic updates prevent corruption
- No complex coordination needed

```python
from atomicio import SafeFile
import time
from pathlib import Path

class ConfigManager:
    """Thread-safe and process-safe configuration manager using SafeFile."""

    def __init__(self, config_path='app_config.json'):
        self.config_file = SafeFile(config_path, timeout=2.0)
        self._ensure_config_exists()

    def _ensure_config_exists(self):
        """Create default config if file doesn't exist."""
        if not Path(self.config_file.path).exists():
            default_config = {
                'app_name': 'MyWebApp',
                'version': '1.0.0',
                'debug': False,
                'max_connections': 100,
                'created_at': time.time(),
                'user_preferences': {}
            }
            self.config_file.write(default_config)

    def get_setting(self, key, default=None):
        """Get a configuration setting safely."""
        config = self.config_file.read() or {}
        return config.get(key, default)

    def update_setting(self, key, value):
        """Update a single configuration setting atomically."""
        config = self.config_file.read() or {}
        config[key] = value
        config['last_modified'] = time.time()
        self.config_file.write(config)

    def update_user_preference(self, user_id, preference_key, value):
        """Update a user-specific preference safely."""
        config = self.config_file.read() or {}

        # Ensure user preferences structure exists
        if 'user_preferences' not in config:
            config['user_preferences'] = {}
        if user_id not in config['user_preferences']:
            config['user_preferences'][user_id] = {}

        # Update the specific preference
        config['user_preferences'][user_id][preference_key] = value
        config['last_modified'] = time.time()

        # Atomic write ensures consistency
        self.config_file.write(config)

    def get_user_preferences(self, user_id):
        """Get all preferences for a specific user."""
        config = self.config_file.read() or {}
        return config.get('user_preferences', {}).get(user_id, {})

    def reset_to_defaults(self):
        """Reset configuration to default values."""
        current_config = self.config_file.read() or {}

        # Preserve user preferences but reset system settings
        default_config = {
            'app_name': 'MyWebApp',
            'version': '1.0.0',
            'debug': False,
            'max_connections': 100,
            'created_at': current_config.get('created_at', time.time()),
            'last_modified': time.time(),
            'user_preferences': current_config.get('user_preferences', {})
        }

        self.config_file.write(default_config)

# Example usage demonstrating thread and process safety
if __name__ == "__main__":
    # This can be safely used across multiple processes/threads
    config_manager = ConfigManager()

    # Safe updates from multiple workers
    config_manager.update_setting('debug', True)
    config_manager.update_setting('max_connections', 200)

    # Safe user preference updates
    config_manager.update_user_preference('user123', 'theme', 'dark')
    config_manager.update_user_preference('user123', 'language', 'es')
    config_manager.update_user_preference('user456', 'theme', 'light')

    # Read operations are always consistent
    debug_mode = config_manager.get_setting('debug')
    user123_prefs = config_manager.get_user_preferences('user123')

    print(f"Debug mode: {debug_mode}")
    print(f"User123 preferences: {user123_prefs}")

    # Demonstrate format persistence
    import json
    with open('app_config.json', 'r') as f:
        print("Raw file content:")
        print(json.dumps(json.load(f), indent=2))
```

**Key Benefits in this Example:**
- ✅ **Atomic Updates**: No corruption even if process crashes during write
- ✅ **Multi-Process Safe**: Web workers can safely update config simultaneously
- ✅ **Auto-Serialization**: JSON format handled automatically
- ✅ **Simple API**: Minimal code required for robust behavior
- ✅ **Consistent Reads**: Always read complete, valid configuration

</details>

---

## 🔗 ThreadedSafeFile - Cross-Operation Coordination

### 🧠 Working Principle

ThreadedSafeFile is designed for applications that need **complex file operations with cross-operation locking**. It provides advanced coordination capabilities where multiple operations must be performed atomically as a sequence.

**Key Concepts:**
- **Per-File Locking**: Each file path gets its own threading.RLock instance
- **Reentrant Locks**: Same thread can acquire the same file lock multiple times
- **Lock Persistence**: The `locked()` context manager holds the lock across multiple operations
- **Cross-Operation Atomicity**: Ensure no other thread can access the file during complex workflows
- **Timeout Control**: Configurable timeouts prevent deadlocks

**When to Use ThreadedSafeFile:**
- ✅ Complex multi-step file operations that must be atomic
- ✅ Data processing pipelines with intermediate steps
- ✅ Applications requiring cross-operation coordination
- ✅ Multi-threaded applications with complex synchronization needs
- ✅ Scenarios where you need to hold a lock while performing multiple operations
- ❌ Simple single-operation file access (use SafeFile)
- ❌ Async applications (use AsyncSafeFile)

### 📋 Complete API Reference

#### **Constructor**
```python
ThreadedSafeFile(path, timeout=True)
```
- `path`: File path (str or Path object)
- `timeout`: Lock acquisition timeout behavior:
  - `True` (default): 15 seconds timeout
  - `False` or `None`: Wait forever (infinite blocking)
  - `int` or `float`: Wait that many seconds (e.g., `5.0`)

**Examples:**
```python
ThreadedSafeFile('data.yaml')              # 15 second timeout (default)
ThreadedSafeFile('data.yaml', timeout=5.0) # 5 second timeout
ThreadedSafeFile('data.yaml', timeout=None) # Wait forever
```

#### **Core Methods**
```python
# Read/Write Operations (each acquires and releases lock)
read() -> Any | None                    # Read and deserialize file content
write(data: Any) -> None               # Write and serialize data atomically
update(data: Any) -> None              # Update existing data (merge for dicts)

# Binary Operations
read_bytes() -> bytes | None           # Read raw bytes
write_bytes(data: bytes) -> None       # Write raw bytes

# Text Operations
append(text: str) -> None              # Append text to file

# Utility Methods
@staticmethod
list_supported_formats() -> list      # List supported file formats
```

#### **Advanced Context Manager**
```python
# Cross-operation locking context
with ThreadedSafeFile('data.yaml', timeout=5.0).locked() as f:
    # Lock is held for the entire block
    data = f.read() or {}

    # Perform complex operations while holding lock
    data['step1'] = process_data(data)
    f.write(data)

    data['step2'] = more_processing(data)
    f.write(data)

    # Lock automatically released at the end
```

#### **Lock Behavior**
- **Individual Operations**: Each `read()`, `write()`, etc. acquires and releases lock
- **Locked Context**: `locked()` holds the lock for the entire context duration
- **Reentrant**: Same thread can call `read()`, `write()` within `locked()` context
- **Per-File**: Different files have independent locks
- **Timeout**: Configurable timeout prevents indefinite blocking

#### **Custom Exceptions**
ThreadedSafeFile raises specific exceptions for different error conditions:
- **`FileReadError`** - When file reading fails (corrupt file, permission issues, etc.)
- **`FileWriteError`** - When file writing fails (disk full, permission denied, etc.)
- **`FileAppendError`** - When appending to file fails (similar to write errors)
- **`AsyncTimeoutError`** - When lock acquisition times out (prevents deadlocks)

### 🎯 Use Case Example: Data Processing Pipeline

<details>
<summary><strong>📖 Click to expand: Complete Data Processing Pipeline Example</strong></summary>

**Scenario**: A multi-threaded data analysis application where multiple workers process large datasets. Each processing stage must be atomic, and intermediate results must be safely stored and passed between stages.

**Why ThreadedSafeFile is Perfect:**
- Complex multi-stage processing requires cross-operation locking
- Multiple threads working on the same dataset files
- Intermediate results must be consistent
- Need to prevent race conditions during processing stages

```python
from atomicio import ThreadedSafeFile
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import List, Dict, Any
import hashlib

@dataclass
class ProcessingJob:
    """Represents a data processing job."""
    job_id: str
    input_data: List[int]
    stages_completed: List[str]
    results: Dict[str, Any]
    created_at: float
    updated_at: float

class DataProcessor:
    """
    Multi-threaded data processor using ThreadedSafeFile for safe coordination.

    This demonstrates ThreadedSafeFile's strength in complex workflows where
    multiple operations must be performed atomically as a sequence.
    """

    def __init__(self, results_file='processing_results.json'):
        self.results_file = ThreadedSafeFile(results_file, timeout=10.0)
        self.active_jobs = ThreadedSafeFile('active_jobs.json', timeout=5.0)
        self._ensure_files_exist()

    def _ensure_files_exist(self):
        """Initialize result files if they don't exist."""
        # Using individual operations for simple initialization
        if self.results_file.read() is None:
            self.results_file.write({'completed_jobs': {}, 'stats': {'total_processed': 0}})

        if self.active_jobs.read() is None:
            self.active_jobs.write({'jobs': {}, 'worker_assignments': {}})

    def submit_job(self, data: List[int]) -> str:
        """Submit a new processing job safely."""
        job_id = hashlib.md5(str(data).encode()).hexdigest()[:8]
        timestamp = time.time()

        job = ProcessingJob(
            job_id=job_id,
            input_data=data,
            stages_completed=[],
            results={},
            created_at=timestamp,
            updated_at=timestamp
        )

        # Use locked context for multi-step job submission
        with self.active_jobs.locked() as f:
            active_data = f.read()

            # Check if job already exists
            if job_id in active_data['jobs']:
                return job_id  # Job already submitted

            # Add new job and update worker assignment
            active_data['jobs'][job_id] = job.__dict__
            active_data['worker_assignments'][job_id] = None  # Not assigned yet

            f.write(active_data)
            print(f"✅ Job {job_id} submitted successfully")

        return job_id

    def process_stage_1_validation(self, job_id: str, worker_id: str) -> bool:
        """Stage 1: Validate input data (requires cross-operation locking)."""
        print(f"🔍 Worker {worker_id} starting Stage 1 (validation) for job {job_id}")

        with self.active_jobs.locked() as f:
            active_data = f.read()

            if job_id not in active_data['jobs']:
                print(f"❌ Job {job_id} not found")
                return False

            job_data = active_data['jobs'][job_id]

            # Assign worker if not already assigned
            if active_data['worker_assignments'][job_id] is None:
                active_data['worker_assignments'][job_id] = worker_id
                print(f"👷 Worker {worker_id} assigned to job {job_id}")
            elif active_data['worker_assignments'][job_id] != worker_id:
                print(f"⚠️  Job {job_id} already assigned to another worker")
                return False

            # Perform validation
            input_data = job_data['input_data']
            if not input_data or len(input_data) == 0:
                print(f"❌ Invalid input data for job {job_id}")
                return False

            # Update job progress (multiple operations under same lock)
            job_data['stages_completed'].append('validation')
            job_data['results']['validation'] = {
                'data_length': len(input_data),
                'data_range': [min(input_data), max(input_data)],
                'validation_time': time.time()
            }
            job_data['updated_at'] = time.time()

            # Save updated job state
            active_data['jobs'][job_id] = job_data
            f.write(active_data)

            print(f"✅ Stage 1 (validation) completed for job {job_id}")
            return True

    def process_stage_2_computation(self, job_id: str, worker_id: str) -> bool:
        """Stage 2: Heavy computation (requires lock persistence)."""
        print(f"🧮 Worker {worker_id} starting Stage 2 (computation) for job {job_id}")

        with self.active_jobs.locked() as f:
            active_data = f.read()

            if job_id not in active_data['jobs']:
                return False

            job_data = active_data['jobs'][job_id]

            # Verify worker assignment and prerequisites
            if active_data['worker_assignments'][job_id] != worker_id:
                print(f"⚠️  Worker {worker_id} not assigned to job {job_id}")
                return False

            if 'validation' not in job_data['stages_completed']:
                print(f"⚠️  Job {job_id} hasn't completed validation stage")
                return False

            # Perform computation while holding lock (simulated work)
            input_data = job_data['input_data']
            print(f"🔄 Computing results for {len(input_data)} data points...")

            # Simulate processing time
            time.sleep(0.5)

            # Heavy computation results
            results = {
                'sum': sum(input_data),
                'average': sum(input_data) / len(input_data),
                'squared_sum': sum(x*x for x in input_data),
                'computation_time': time.time()
            }

            # Update job with computation results
            job_data['stages_completed'].append('computation')
            job_data['results']['computation'] = results
            job_data['updated_at'] = time.time()

            active_data['jobs'][job_id] = job_data
            f.write(active_data)

            print(f"✅ Stage 2 (computation) completed for job {job_id}")
            return True

    def process_stage_3_finalization(self, job_id: str, worker_id: str) -> bool:
        """Stage 3: Finalize and move to completed (complex cross-file operation)."""
        print(f"🏁 Worker {worker_id} starting Stage 3 (finalization) for job {job_id}")

        # This demonstrates the power of ThreadedSafeFile: coordinating multiple files
        with self.active_jobs.locked() as active_f:
            active_data = active_f.read()

            if job_id not in active_data['jobs']:
                return False

            job_data = active_data['jobs'][job_id]

            # Verify prerequisites
            required_stages = ['validation', 'computation']
            if not all(stage in job_data['stages_completed'] for stage in required_stages):
                print(f"⚠️  Job {job_id} missing required stages")
                return False

            # Finalize results
            final_results = {
                'job_id': job_id,
                'input_length': len(job_data['input_data']),
                'validation_results': job_data['results']['validation'],
                'computation_results': job_data['results']['computation'],
                'total_processing_time': time.time() - job_data['created_at'],
                'completed_at': time.time(),
                'processed_by': worker_id
            }

            # Remove from active jobs
            del active_data['jobs'][job_id]
            del active_data['worker_assignments'][job_id]
            active_f.write(active_data)

        # Now update completed jobs (separate file operation)
        with self.results_file.locked() as results_f:
            results_data = results_f.read()

            # Add to completed jobs
            results_data['completed_jobs'][job_id] = final_results
            results_data['stats']['total_processed'] += 1

            results_f.write(results_data)

            print(f"🎉 Job {job_id} completed and finalized!")
            return True

    def process_job_complete_pipeline(self, job_id: str, worker_id: str) -> bool:
        """Process a complete job through all stages."""
        try:
            # Each stage can be called independently or as part of a pipeline
            if not self.process_stage_1_validation(job_id, worker_id):
                return False

            if not self.process_stage_2_computation(job_id, worker_id):
                return False

            if not self.process_stage_3_finalization(job_id, worker_id):
                return False

            return True

        except Exception as e:
            print(f"❌ Error processing job {job_id}: {e}")
            return False

    def get_processing_status(self) -> Dict[str, Any]:
        """Get current processing status (demonstrates safe read operations)."""
        active_data = self.active_jobs.read() or {'jobs': {}, 'worker_assignments': {}}
        results_data = self.results_file.read() or {'completed_jobs': {}, 'stats': {'total_processed': 0}}

        return {
            'active_jobs': len(active_data['jobs']),
            'completed_jobs': len(results_data['completed_jobs']),
            'total_processed': results_data['stats']['total_processed'],
            'worker_assignments': active_data['worker_assignments']
        }

# Example usage demonstrating complex multi-threaded processing
if __name__ == "__main__":
    processor = DataProcessor()

    # Submit multiple jobs
    jobs = []
    test_datasets = [
        [1, 2, 3, 4, 5],
        [10, 20, 30, 40, 50, 60],
        [100, 200, 300],
        [1, 3, 5, 7, 9, 11, 13, 15]
    ]

    for i, dataset in enumerate(test_datasets):
        job_id = processor.submit_job(dataset)
        jobs.append(job_id)
        print(f"Submitted job {i+1}: {job_id}")

    # Process jobs with multiple workers
    def worker_task(worker_id, job_ids):
        """Worker function that processes assigned jobs."""
        results = []
        for job_id in job_ids:
            print(f"\n🚀 Worker {worker_id} processing job {job_id}")
            success = processor.process_job_complete_pipeline(job_id, worker_id)
            results.append((job_id, success))
            time.sleep(0.1)  # Brief pause between jobs
        return results

    # Run with multiple workers
    with ThreadPoolExecutor(max_workers=3) as executor:
        # Distribute jobs among workers
        worker_jobs = [jobs[i::3] for i in range(3)]  # Round-robin distribution

        futures = []
        for i, job_list in enumerate(worker_jobs):
            if job_list:  # Only submit if worker has jobs
                future = executor.submit(worker_task, f"Worker-{i+1}", job_list)
                futures.append(future)

        # Wait for all workers to complete
        all_results = []
        for future in as_completed(futures):
            worker_results = future.result()
            all_results.extend(worker_results)
            print(f"Worker completed {len(worker_results)} jobs")

    # Show final status
    print("\n📊 Final Processing Status:")
    status = processor.get_processing_status()
    for key, value in status.items():
        print(f"  {key}: {value}")

    # Demonstrate successful coordination
    if status['active_jobs'] == 0 and status['completed_jobs'] == len(jobs):
        print("\n🎉 All jobs processed successfully with perfect coordination!")
    else:
        print(f"\n⚠️  Some jobs may have failed. Check the logs.")
```

**Key Benefits in this Example:**
- ✅ **Cross-Operation Atomicity**: Multiple file operations within `locked()` contexts are atomic
- ✅ **Perfect Coordination**: No race conditions between workers processing the same files
- ✅ **Complex Workflows**: Multi-stage processing with proper synchronization
- ✅ **Reentrant Safety**: Workers can call methods that internally use locks
- ✅ **Timeout Protection**: Configurable timeouts prevent deadlocks
- ✅ **Per-File Granularity**: Different files can be accessed independently

</details>

---

## ⚡ AsyncSafeFile - Async-Native Operations

### 🧠 Working Principle

AsyncSafeFile is designed specifically for **asyncio-based applications** that need thread-safe file operations without blocking the event loop. It provides native async support with proper async context managers and lock coordination.

**Key Concepts:**
- **Async-Native**: All operations are `async`/`await` compatible
- **Per-File AsyncLock**: Each file path gets its own `asyncio.Lock` instance
- **Non-Blocking**: Designed to work seamlessly with asyncio event loops
- **Async Context Managers**: `async with asf.locked()` for cross-operation coordination
- **Async Timeout Control**: Proper async timeout handling with task cancellation
- **Event Loop Safe**: All operations are safe within asyncio applications

**When to Use AsyncSafeFile:**
- ✅ Asyncio-based applications (FastAPI, aiohttp, etc.)
- ✅ Async web servers and APIs
- ✅ Async data processing pipelines
- ✅ Applications using async/await patterns
- ✅ Non-blocking file operations in event loops
- ❌ Synchronous applications (use SafeFile or ThreadedSafeFile)
- ❌ Simple scripts without async requirements

### 📋 Complete API Reference

#### **Constructor**
```python
AsyncSafeFile(path, timeout=True)
```
- `path`: File path (str or Path object)
- `timeout`: Lock acquisition timeout behavior:
  - `True` (default): 15 seconds timeout
  - `False` or `None`: Wait forever (infinite blocking)
  - `int` or `float`: Wait that many seconds (e.g., `5.0`)

**Examples:**
```python
AsyncSafeFile('data.json')              # 15 second timeout (default)
AsyncSafeFile('data.json', timeout=5.0) # 5 second timeout
AsyncSafeFile('data.json', timeout=None) # Wait forever
```

#### **Core Async Methods**
```python
# Read/Write Operations (all async)
async read() -> Any | None              # Read and deserialize file content
async write(data: Any) -> None          # Write and serialize data atomically
async update(data: Any) -> None         # Update existing data (merge for dicts)

# Binary Operations
async read_bytes() -> bytes | None      # Read raw bytes
async write_bytes(data: bytes) -> None  # Write raw bytes

# Text Operations
async append(text: str) -> None         # Append text to file

# Utility Methods
@staticmethod
list_supported_formats() -> list       # List supported file formats (sync)
```

#### **Async Context Manager**
```python
# Cross-operation async locking context
async with AsyncSafeFile('data.json', timeout=5.0).locked() as f:
    # Async lock is held for the entire block
    data = await f.read() or {}

    # Perform complex async operations while holding lock
    data['step1'] = await async_process_data(data)
    await f.write(data)

    data['step2'] = await more_async_processing(data)
    await f.write(data)

    # Lock automatically released at the end
```

#### **Async Lock Behavior**
- **Individual Operations**: Each `await read()`, `await write()`, etc. acquires and releases lock
- **Locked Context**: `async with locked()` holds the lock for the entire context duration
- **Async-Safe**: Designed for asyncio event loops and coroutines
- **Per-File**: Different files have independent async locks
- **Timeout**: Async timeout with proper task cancellation

#### **Custom Exceptions**
AsyncSafeFile raises specific exceptions for different error conditions:
- **`FileReadError`** - When file reading fails (corrupt file, permission issues, etc.)
- **`FileWriteError`** - When file writing fails (disk full, permission denied, etc.)
- **`FileAppendError`** - When appending to file fails (similar to write errors)
- **`AsyncTimeoutError`** - When async operations timeout (includes timeout details)
- **`AsyncLockError`** - When async lock operations fail unexpectedly

### 🎯 Use Case Example: Async Web API with Data Processing

<details>
<summary><strong>📖 Click to expand: Complete Async Web API Example</strong></summary>

**Scenario**: A FastAPI-based web service that processes user uploads, analyzes data asynchronously, and stores results. Multiple async workers handle requests concurrently, and file operations must be coordinated without blocking the event loop.

**Why AsyncSafeFile is Perfect:**
- FastAPI async endpoints require non-blocking operations
- Multiple concurrent requests processing the same data files
- Need async coordination for complex data processing workflows
- Event loop must never be blocked by file I/O

```python
from atomicio import AsyncSafeFile
import asyncio
import time
import aiohttp
import aiofiles
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from fastapi import FastAPI, UploadFile, HTTPException
import uvicorn
import hashlib
import json
from pathlib import Path

@dataclass
class AnalysisTask:
    """Represents an async data analysis task."""
    task_id: str
    filename: str
    status: str  # 'pending', 'processing', 'completed', 'failed'
    created_at: float
    st