Threadsafe and Fault-Tolerant File Writes in Python: Fixing Temp File Cleanup, Concurrency, and Reusability Issues

import tempfile
 
# Creates a temp file, but what if the script crashes before deletion?
temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False)  # delete=False leaves it behind!
temp_file.write("sensitive data")
# Oops! If the script crashes here, temp_file is never closed or deleted.
temp_file.close()
os.unlink(temp_file.name)  # Too late if crash happens before this line!

import tempfile
 
# Use a context manager to auto-close/delete the temp file
with tempfile.NamedTemporaryFile(mode='w', delete=True) as temp_file:
    temp_file.write("data to stage")
    # File is closed and deleted automatically when exiting the 'with' block

import tempfile
import os
 
temp_file = None
try:
    temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False)
    temp_file.write("data")
finally:
    if temp_file:
        temp_file.close()
        os.unlink(temp_file.name)  # Ensure deletion even if an error occurs

with tempfile.TemporaryDirectory() as temp_dir:
    file1 = os.path.join(temp_dir, "file1.txt")
    with open(file1, 'w') as f:
        f.write("batch data 1")
    # All files in temp_dir are deleted when exiting the 'with' block

import atexit
import tempfile
import os
 
temp_file = tempfile.NamedTemporaryFile(mode='w', delete=False)
atexit.register(lambda: os.unlink(temp_file.name) if os.path.exists(temp_file.name) else None)
 
# ... rest of the program ...

import threading
 
class ThreadsafeFileAppender:
    def __init__(self, file_path):
        self.file_path = file_path
        self.lock = threading.Lock()  # Serialize access
 
    def append(self, data):
        with self.lock:  # Only one thread enters this block at a time
            with open(self.file_path, 'a') as f:
                f.write(data + '\n')
 
# Usage:
appender = ThreadsafeFileAppender("logs.txt")
 
def worker(data):
    appender.append(data)
 
# Start 10 threads writing to logs.txt safely
threads = [threading.Thread(target=worker, args=(f"Thread {i}",)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

import threading
import queue
 
class QueueBasedFileWriter:
    def __init__(self, file_path, max_queue_size=1000):
        self.file_path = file_path
        self.queue = queue.Queue(maxsize=max_queue_size)
        self.writer_thread = threading.Thread(target=self._write_loop, daemon=True)
        self.writer_thread.start()
 
    def _write_loop(self):
        """Dedicated thread to dequeue and write data."""
        with open(self.file_path, 'a') as f:
            while True:
                data = self.queue.get()  # Blocks until data is available
                f.write(data + '\n')
                self.queue.task_done()  # Notify queue that task is complete
 
    def write(self, data):
        """Enqueue data; non-blocking (or blocks if queue is full)."""
        self.queue.put(data)
 
# Usage:
writer = QueueBasedFileWriter("high_volume_logs.txt")
for i in range(1000):
    writer.write(f"Log entry {i}")
writer.queue.join()  # Wait for all enqueued data to be written

import os
import tempfile
import hashlib
 
def atomic_write(target_path, data):
    # Create a temp file in the same directory as target_path
    temp_dir = os.path.dirname(target_path)
    with tempfile.NamedTemporaryFile(
        mode='w', 
        dir=temp_dir, 
        delete=False,  # Don't auto-delete; we'll rename it
        suffix='.tmp'
    ) as temp_file:
        temp_file.write(data)
        temp_path = temp_file.name  # Path to the temp file
 
    try:
        # Optional: Verify the temp file's content
        with open(temp_path, 'r') as f:
            content = f.read()
            if hashlib.sha256(content.encode()).hexdigest() != expected_checksum:
                raise ValueError("Data corruption detected!")
 
        # Atomic rename: Replace target_path with temp_path
        os.rename(temp_path, target_path)
    except Exception as e:
        os.unlink(temp_path)  # Clean up on failure
        raise e

import logging
from tenacity import retry, stop_after_attempt, wait_exponential  # Use 'tenacity' for retries
 
logging.basicConfig(level=logging.INFO)
 
@retry(
    stop=stop_after_attempt(3),  # Max 3 retries
    wait=wait_exponential(multiplier=1, min=2, max=10)  # Backoff: 2s, 4s, 8s
)
def atomic_write_with_retry(target_path, data):
    try:
        atomic_write(target_path, data)
    except OSError as e:
        logging.warning(f"Write failed: {e}. Retrying...")
        raise  # Let tenacity handle retry

import os
import tempfile
import threading
import logging
from contextlib import contextmanager
 
class ThreadsafeFileWriter:
    def __init__(self, target_path, max_retries=3):
        self.target_path = target_path
        self.lock = threading.Lock()  # For thread safety
        self.max_retries = max_retries
        self.logger = logging.getLogger(__name__)
 
    @contextmanager
    def _temp_file(self):
        """Context manager for temp file in the target directory."""
        temp_dir = os.path.dirname(self.target_path)
        with tempfile.NamedTemporaryFile(
            mode='w', 
            dir=temp_dir, 
            delete=False, 
            suffix='.tmp'
        ) as temp_file:
            try:
                yield temp_file
            finally:
                temp_file.close()
                if os.path.exists(temp_file.name):
                    os.unlink(temp_file.name)  # Cleanup if context exits early
 
    def write(self, data):
        """Thread-safe atomic write."""
        with self.lock:  # Ensure only one thread writes at a time
            for attempt in range(self.max_retries):
                try:
                    with self._temp_file() as temp_file:
                        temp_file.write(data)
                        temp_path = temp_file.name
 
                    # Rename temp file to target (atomic)
                    os.rename(temp_path, self.target_path)
                    self.logger.info("Write successful.")
                    return
                except Exception as e:
                    self.logger.warning(f"Attempt {attempt+1} failed: {e}")
                    if attempt == self.max_retries - 1:
                        raise  # Re-raise after final attempt

import threading
import time
import logging
 
# Configure logging
logging.basicConfig(level=logging.INFO)
 
def thread_task(writer, thread_id):
    """Simulate a thread writing logs."""
    for i in range(3):  # Write 3 entries per thread
        log_entry = f"Thread {thread_id}, Entry {i}: {time.time()}"
        writer.write(log_entry)
        time.sleep(0.1)  # Simulate work
 
if __name__ == "__main__":
    target_file = "multi_thread_logs.txt"
    writer = ThreadsafeFileWriter(target_file)
 
    # Start 5 threads
    threads = [
        threading.Thread(target=thread_task, args=(writer, i))
        for i in range(5)
    ]
 
    for t in threads:
        t.start()
    for t in threads:
        t.join()
 
    # Verify the output
    with open(target_file, 'r') as f:
        print("Final log content:")
        print(f.read())