Custom Context Managers

class AtomicFileSwap:
    def __init__(self, filename):
        self.filename = filename
        self.temp_filename = f"{filename}.tmp"
        self.committed = False

    def __enter__(self):
        print(f"[SETUP] Preparing atomic write for {self.filename}")
        self.file = open(self.temp_filename, "w", encoding="utf-8")
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        if exc_type is None:
            # Success: atomically replace file
            import os
            os.replace(self.temp_filename, self.filename)
            self.committed = True
            print(f"[SUCCESS] {self.filename} updated atomically")
            return False
        else:
            # Failure: discard temporary file
            import os
            if os.path.exists(self.temp_filename):
                os.remove(self.temp_filename)
            print(f"[ROLLBACK] Discarded temporary file due to {exc_type.__name__}")
            return False  # Propagate exception

# Simulating clean execution
with AtomicFileSwap("app_config.json") as f:
    f.write('{"status": "active"}')

class IgnoreExceptions:
    def __init__(self, *ignorable):
        self.ignorable = ignorable

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None and issubclass(exc_type, self.ignorable):
            print(f"[SUPPRESSED] Safely caught and suppressed {exc_type.__name__}: {exc_val}")
            return True  # Suppress!
        return False     # Propagate unexpected errors!

# 1. Ignored error: suppresses FileNotFoundError
with IgnoreExceptions(FileNotFoundError, KeyError):
    print("Attempting risky operation...")
    raise FileNotFoundError("Missing cache file")

print("Code successfully resumed after suppressed error!")

from contextlib import ExitStack, contextmanager

@contextmanager
def mock_lock(resource_id):
    print(f"Acquiring lock: {resource_id}")
    try:
        yield resource_id
    finally:
        print(f"Releasing lock: {resource_id}")

def synchronize_nodes(node_ids):
    with ExitStack() as stack:
        # Dynamically acquire an arbitrary list of locks
        locks = [stack.enter_context(mock_lock(n)) for n in node_ids]
        # Register an arbitrary cleanup callback
        stack.callback(print, "All nodes synchronized successfully")
        print("Active locks held:", locks)

synchronize_nodes(["NODE_A", "NODE_B", "NODE_C"])

class Timer:
    def __enter__(self):
        print("Timer started")
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Timer stopped")

with Timer():
    print("Running critical task...")

# Service integration layer
def handle_request(payload: dict) -> dict:
    return {"status": "success", "processed": True}

# High-resilience production pattern
import time

def process_stream(events: list) -> int:
    processed = 0
    for event in events:
        if event:
            processed += 1
    return processed