Project: Resource Manager with Context Manager0%
Itertools for Iteration Tools

Project: Resource Manager with Context Manager

Beginner12 min read•Updated: Jul 11, 2026
Study Materials

Project: Resource Manager with Context Manager

In distributed computing and enterprise backend systems, safely managing critical external resources—such as file locks, scratch workspaces, database connections, and transactional state—is essential to prevent data corruption and resource leaks.

In this project, we will design and construct a production-ready Atomic File Transaction & Workspace Manager. It guarantees ACID-like atomicity for file updates: modifications are executed within an isolated staging environment and committed atomically only upon successful completion. If an error occurs, modifications are automatically rolled back, leaving the original resources untouched.

1. Project Requirements & Architecture

The Resource Manager must fulfill the following operational criteria:

  1. 1
    Isolation: All write operations occur on a temporary staging buffer rather than directly mutating the target file.
  2. 2
    Atomicity: The target file is updated via an atomic filesystem swap (os.replace) only when the with block finishes cleanly without exceptions.
  3. 3
    Rollback Guarantee: In the event of any failure (validation error, network abort, or disk error), all staged artifacts are purged, preserving the original file.
  4. 4
    Context Protocol Compliance: Fully implement __enter__ and __exit__, handling exception inspection and suppression policies.
  5. 5
    Telemetry & Audit Trail: Track execution latency, bytes written, and operational status.
Output
Atomic Transaction Lifecycle
│
with AtomicFileWriter("data.json") as f:
│
▼
Create Staging File: data.json.tmp
│
▼
Yield Staging File Descriptor
│
▼
User Performs Writes to Staging
│
┌──────────────┴──────────────┐
▼ ▼
Success / Clean Exit Exception / Failure
│ │
▼ ▼
Flush & Sync to Disk Close Staging File
│ │
▼ ▼
Atomic OS Replace Delete Staging File
(data.json.tmp -> data.json) │
│ ▼
▼ Original File Untouched
Target Updated Atomically Re-raise Exception

2. Production Implementation

Python
import os
import shutil
import tempfile
import time
from typing import IO, Optional, Type
from types import TracebackType
 
class AtomicFileWriter:
"""A context manager that guarantees atomic write operations to a file.
All writes are directed to a temporary staging file located in the same
directory as the destination. On clean completion, an atomic filesystem
replacement replaces the original target with the staging file.
"""
 
def __init__(self, target_path: str, mode: str = "w", encoding: str = "utf-8") -> None:
self.target_path = os.path.abspath(target_path)
self.mode = mode
self.encoding = encoding
self._dir = os.path.dirname(self.target_path)
self._temp_path: Optional[str] = None
self._file_handle: Optional[IO] = None
self.bytes_written = 0
self.start_time = 0.0
 
def __enter__(self) -> IO:
"""Initializes the staging file and returns the open file handle."""
os.makedirs(self._dir, exist_ok=True)
self.start_time = time.perf_counter()
 
# Create temporary file in the same filesystem/directory to guarantee atomic rename
prefix = f".{os.path.basename(self.target_path)}.tmp_"
temp_fd, self._temp_path = tempfile.mkstemp(prefix=prefix, dir=self._dir)
# Open Python stream wrapper around raw file descriptor
self._file_handle = open(temp_fd, mode=self.mode, encoding=self.encoding)
print(f"[STAGE] Created temporary staging buffer: {os.path.basename(self._temp_path)}")
return self._file_handle
 
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType]
) -> bool:
"""Finalizes the transaction: commits on success, purges on failure."""
elapsed_ms = (time.perf_counter() - self.start_time) * 1000
 
# Close the open staging file handle
if self._file_handle and not self._file_handle.closed:
self._file_handle.flush()
os.fsync(self._file_handle.fileno()) # Force write to physical storage
self._file_handle.close()
 
if exc_type is not None:
# Failure scenario: Roll back modifications
print(f"[ROLLBACK] Exception detected: {exc_val}")
if self._temp_path and os.path.exists(self._temp_path):
os.remove(self._temp_path)
print(f"[ROLLBACK] Cleaned up staging file. Target remained untouched.")
print(f"[METRICS] Aborted transaction after {elapsed_ms:.2f} ms")
return False # Propagate exception to caller
 
# Success scenario: Commit staging file to target atomically
if self._temp_path and os.path.exists(self._temp_path):
os.replace(self._temp_path, self.target_path)
print(f"[COMMIT] Atomically swapped staging buffer into: {self.target_path}")
print(f"[METRICS] Transaction committed successfully in {elapsed_ms:.2f} ms")
 
return True

3. Alternative Implementation Using contextlib

For lightweight transactional scopes, the same architectural pattern can be expressed succinctly using @contextlib.contextmanager:

Python
import contextlib
import os
import tempfile
from typing import Generator, IO
 
@contextlib.contextmanager
def atomic_write(filepath: str, mode: str = "w", encoding: str = "utf-8") -> Generator[IO, None, None]:
"""Generator-based atomic file writer using contextlib."""
abs_path = os.path.abspath(filepath)
directory = os.path.dirname(abs_path)
os.makedirs(directory, exist_ok=True)
fd, temp_path = tempfile.mkstemp(prefix=".tmp_write_", dir=directory)
handle = open(fd, mode=mode, encoding=encoding)
try:
yield handle
handle.flush()
os.fsync(handle.fileno())
handle.close()
os.replace(temp_path, abs_path)
print(f"[CONTEXTLIB] Successfully committed: {abs_path}")
except BaseException as err:
handle.close()
if os.path.exists(temp_path):
os.remove(temp_path)
print(f"[CONTEXTLIB] Aborted and purged temporary file due to: {err}")
raise

4. Verification and Demonstration

Python
# -------------------------------------------------------------
# Demonstration Scenario 1: Successful Atomic Commit
# -------------------------------------------------------------
target_file = "production_config.json"
 
# Write initial baseline file
with open(target_file, "w") as f:
f.write('{"status": "INITIAL_VERSION", "cluster_size": 3}')
 
print("--- Scenario 1: Successful Update ---")
with AtomicFileWriter(target_file) as writer:
writer.write('{"status": "UPDATED_VERSION", "cluster_size": 10}')
 
with open(target_file, "r") as f:
print(f"File content after success: {f.read()}\n")
 
# -------------------------------------------------------------
# Demonstration Scenario 2: Rollback on Error
# -------------------------------------------------------------
print("--- Scenario 2: Error During Write (Automatic Rollback) ---")
try:
with AtomicFileWriter(target_file) as writer:
writer.write('{"status": "CORRUPTED_INCOMPLETE_PAYLOAD"')
# Simulate unexpected application failure mid-write
raise ValueError("Invalid payload: schema validation failed!")
except ValueError as err:
print(f"Caught error in caller: {err}")
 
# Verify that original file was not corrupted
with open(target_file, "r") as f:
print(f"File content after rollback: {f.read()}")
 
# Clean up test artifact
if os.path.exists(target_file):
os.remove(target_file)

5. Architectural Key Takeaways

  1. 1
    Atomic File Swaps: The os.replace system call is atomic on POSIX and modern Windows systems provided both the source and target reside on the same filesystem mount.
  2. 2
    Deterministic Cleanup: By utilizing __exit__ and os.fsync, in-flight buffers are forced onto non-volatile storage before replacement occurs, avoiding partially written files during system power loss.
  3. 3
    Separation of Concerns: Business code focuses exclusively on writing data, while the context manager guarantees transactional safety and error recovery.

Multiple Choice Questions

1.

Why must the temporary staging file in an atomic file writer be created on the same filesystem directory as the destination file? A. Because Python cannot create files in other directories. B. Because atomic replacement operations like os.replace() require both paths to reside on the same filesystem/drive. C. To save disk space. D. Because Windows does not support paths longer than 8 characters.

Answer: B
Explanation:In both POSIX (rename) and Windows (SetFileInformationByHandle), atomic filesystem swaps only work if both files reside on the same physical volume/mount. Cross-filesystem operations require copy-and-delete, which is not atomic.

2.

What is the purpose of calling os.fsync(file.fileno()) before renaming the staging file? A. It compresses the file using gzip. B. It flushes operating system kernel write buffers directly to physical non-volatile storage, preventing zero-length files on sudden power loss. C. It verifies the syntax of the JSON payload. D. It encrypts the file on disk.

Answer: B
Explanation:os.fsync() forces the operating system kernel to flush write caches to physical disk sectors, guaranteeing data integrity before the atomic swap takes place.

3.

In AtomicFileWriter.__exit__, what does returning False when exc_type is not None accomplish? A. It suppresses the exception and continues execution. B. It instructs Python to re-raise the exception so the caller is notified of the failure. C. It deletes the destination file completely. D. It restarts the Python interpreter.

Answer: B
Explanation:Returning False (or None) from __exit__ instructs the runtime to propagate the caught exception up the call stack, ensuring the caller is aware that an error occurred.

4.

What occurs to the temporary staging file if an exception is raised inside the with AtomicFileWriter(...) block? A. It is permanently retained on disk. B. It is automatically closed and deleted via os.remove() in __exit__. C. It replaces the original target file anyway. D. It is renamed to .corrupt.

Answer: B
Explanation:The error-handling logic in __exit__ detects the exception, closes the handle, and removes the staging file, ensuring that no partially written temporary artifacts linger.

5.

Which standard library module provides the mkstemp function used to securely create uniquely named temporary files? A. os B. tempfile C. sys D. contextlib

Answer: B
Explanation:The tempfile module provides tempfile.mkstemp(), which generates uniquely named temporary files safely without race conditions.
Next Lesson

Itertools for Iteration Tools

Continue learning with hands-on practice, examples, and exercises in the upcoming topic.

Related Lessons

Previous LessonNext Lesson
Using contextlibItertools for Iteration Tools

Practice Quiz

Test your understanding of this lesson with 5 questions. Each question has one correct answer.

PreviousNext
PrevNext