Python Context Managers

# ❌ MASALAH 1: File tidak ditutup saat error
f = open("data.txt", "r")
data = f.read()
process(data)  # Jika error di sini, file TIDAK ditutup!
f.close()

# ❌ MASALAH 2: Lupa menutup file
f = open("data.txt", "r")
data = f.read()
# f.close()  — lupa! File tetap terbuka

# ✅ SOLUSI: Menggunakan with statement
with open("data.txt", "r") as f:
    data = f.read()
    process(data)
# File OTOMATIS ditutup, bahkan jika terjadi error

# ❌ MASALAH 3: Lock tidak di-release
lock = threading.Lock()
lock.acquire()
do_work()
# Jika error, lock tidak pernah di-release!
lock.release()

# ✅ SOLUSI dengan context manager
lock = threading.Lock()
with lock:
    do_work()
# Lock otomatis di-release

from pathlib import Path
import threading
import time
import decimal
from decimal import Decimal

# ===== 1. File I/O =====
with open("data.txt", "w") as f:
    f.write("Hello, World!")
# File otomatis ditutup

# Multiple files
with open("input.txt", "r") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

# ===== 2. Threading Lock =====
lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:
        counter += 1

# ===== 3. Decimal Context =====
with decimal.localcontext() as ctx:
    ctx.prec = 50
    result = Decimal(1) / Decimal(7)
    print(f"High precision: {result}")

# Kembali ke default precision
print(f"Default: {Decimal(1) / Decimal(7)}")

# ===== 4. Temporary Directory =====
import tempfile

with tempfile.TemporaryDirectory() as tmpdir:
    tmp_path = Path(tmpdir)
    (tmp_path / "test.txt").write_text("data")
    print(f"Temp dir: {tmp_path}")
# Directory dan semua isinya otomatis dihapus

# ===== 5. Timer Context Manager =====
import contextlib

@contextlib.contextmanager
def timer(label: str):
    start = time.time()
    yield
    elapsed = time.time() - start
    print(f"⏱️ {label}: {elapsed:.3f}s")

with timer("Proses data"):
    time.sleep(1.5)
    total = sum(range(1_000_000))

# ===== 6. Suppress Exceptions =====
with contextlib.suppress(FileNotFoundError):
    Path("nonexistent.txt").unlink()
# FileNotFoundError diabaikan, program lanjut

import contextlib

# ===== Multiple with (Python 3.10+) =====
# with (
#     open("a.txt") as f1,
#     open("b.txt") as f2,
#     open("c.txt") as f3,
# ):
#     pass

# ===== Backward-compatible syntax =====
with open("a.txt") as f1, open("b.txt") as f2:
    data1 = f1.read()
    data2 = f2.read()

# ===== Nested with (Python < 3.10) =====
with open("a.txt") as f1:
    with open("b.txt") as f2:
        with open("c.txt") as f3:
            data = f1.read() + f2.read() + f3.read()

# ===== ExitStack — dynamic number of CM =====
with contextlib.ExitStack() as stack:
    files = [
        stack.enter_context(open(f"file_{i}.txt", "w"))
        for i in range(5)
    ]
    for i, f in enumerate(files):
        f.write(f"Data file {i}\n")
# Semua 5 file otomatis ditutup

# ExitStack dengan conditional context managers
with contextlib.ExitStack() as stack:
    log_file = stack.enter_context(open("app.log", "a"))
    if debug_mode := True:
        debug_file = stack.enter_context(open("debug.log", "a"))

    log_file.write("App started\n")
    # debug_file hanya di-close jika dibuka

class SimpleContextManager:
    """Contoh sederhana context manager."""

    def __init__(self, name: str):
        self.name = name
        print(f"  __init__: {name} dibuat")

    def __enter__(self):
        """Dipanggil saat masuk ke with block.
        Return value akan di-assign ke 'as' variable."""
        print(f"  __enter__: {self.name} masuk")
        return self  # Biasanya return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Dipanggil saat keluar dari with block.
        Parameters:
          exc_type:  Tipe exception (None jika tidak ada error)
          exc_val:   Instance exception
          exc_tb:    Traceback object
        Return True untuk suppress exception, False untuk propagate."""
        if exc_type is not None:
            print(f"  __exit__: Error! {exc_type.__name__}: {exc_val}")
        else:
            print(f"  __exit__: {self.name} selesai tanpa error")
        return False  # Jangan suppress exception

# Menggunakan
print("=== Normal Flow ===")
with SimpleContextManager("resource") as cm:
    print(f"  Body: menggunakan {cm.name}")

print("\n=== With Exception ===")
try:
    with SimpleContextManager("resource") as cm:
        print(f"  Body: mulai...")
        raise ValueError("Something went wrong!")
except ValueError:
    print("  Exception ditangkap di luar with")

class ErrorSuppressor:
    """Context manager yang menekan exception tertentu."""

    def __init__(self, *exceptions):
        self.exceptions = exceptions

    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.exceptions):
            print(f"  ⚠️ Suppressed: {exc_type.__name__}: {exc_val}")
            return True  # Suppress exception!
        return False  # Propagate exception lainnya

# Suppress FileNotFoundError saja
with ErrorSuppressor(FileNotFoundError):
    open("nonexistent.txt")
print("  Program lanjut!")

# Suppress multiple exception types
with ErrorSuppressor(FileNotFoundError, PermissionError):
    open("secret.txt")
print("  Program lanjut juga!")

# ValueError TIDAK di-suppress
try:
    with ErrorSuppressor(FileNotFoundError):
        int("not_a_number")
except ValueError:
    print("  ValueError tetap propagate!")

class RetryContextManager:
    """Context manager yang me-retry saat gagal."""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.attempt = 0

    def __enter__(self):
        self.attempt += 1
        print(f"  Attempt {self.attempt}/{self.max_retries}")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            if self.attempt < self.max_retries:
                print(f"  Retry karena {exc_type.__name__}")
                return True  # Suppress dan akan masuk lagi
            print(f"  Gagal setelah {self.max_retries} attempts")
            return False  # Propagate
        return False

import sqlite3
from typing import Optional

class DatabaseConnection:
    """Context manager untuk database connection."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection: Optional[sqlite3.Connection] = None
        self.cursor: Optional[sqlite3.Cursor] = None

    def __enter__(self):
        print(f"  📦 Connecting to {self.db_path}")
        self.connection = sqlite3.connect(self.db_path)
        self.cursor = self.connection.cursor()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            print(f"  ❌ Error: {exc_val}, rolling back...")
            self.connection.rollback()
        else:
            print(f"  ✅ Committing changes...")
            self.connection.commit()

        self.cursor.close()
        self.connection.close()
        print(f"  🔒 Connection closed")
        return False

    def execute(self, query: str, params: tuple = ()):
        self.cursor.execute(query, params)
        return self.cursor

    def fetchall(self):
        return self.cursor.fetchall()

# Menggunakan
with DatabaseConnection(":memory:") as db:
    db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    db.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    db.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

    db.execute("SELECT * FROM users")
    rows = db.fetchall()
    for row in rows:
        print(f"  User: {row}")

# Dengan error — otomatis rollback
try:
    with DatabaseConnection(":memory:") as db:
        db.execute("CREATE TABLE test (id INTEGER)")
        db.execute("INSERT INTO test VALUES (1)")
        raise RuntimeError("Simulasi error!")
except RuntimeError:
    print("  Error ditangkap, tapi DB sudah di-cleanup!")

import time
import cProfile
import pstats
from typing import Optional

class Timer:
    """Context manager untuk mengukur waktu eksekusi."""

    def __init__(self, label: str = "Block"):
        self.label = label
        self.elapsed: float = 0

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, *args):
        self.elapsed = time.perf_counter() - self.start
        print(f"  ⏱️ {self.label}: {self.elapsed:.4f}s")
        return False

class Profiler:
    """Context manager untuk profiling kode."""

    def __init__(self, top_n: int = 10):
        self.top_n = top_n
        self.profiler = cProfile.Profile()
        self.stats: Optional[pstats.Stats] = None

    def __enter__(self):
        self.profiler.enable()
        return self

    def __exit__(self, *args):
        self.profiler.disable()
        self.stats = pstats.Stats(self.profiler)
        self.stats.sort_stats('cumulative')
        self.stats.print_stats(self.top_n)
        return False

class IndentPrinter:
    """Context manager untuk indented output."""

    def __init__(self, prefix: str = ""):
        self.prefix = prefix
        self._indent = 0

    def __enter__(self):
        self._indent += 2
        print(f"{self.prefix}{{")
        return self

    def __exit__(self, *args):
        self._indent -= 2
        print(f"{self.prefix}}}")
        return False

    def print(self, msg: str):
        print(f"{' ' * self._indent}{msg}")

# Menggunakan Timer
with Timer("Data Processing"):
    data = [x ** 2 for x in range(1_000_000)]

# Menggunakan Profiler
with Profiler(top_n=5):
    data = sorted([x ** 2 for x in range(100_000)])

from contextlib import contextmanager
import time
import os
import sys
from pathlib import Path

# ===== Basic @contextmanager =====
@contextmanager
def managed_resource(name: str):
    """Simple resource manager."""
    print(f"  🔓 Acquiring {name}")
    try:
        yield name  # Nilai ini di-assign ke 'as' variable
    except Exception as e:
        print(f"  ❌ Error in {name}: {e}")
        raise  # Re-raise exception
    finally:
        print(f"  🔒 Releasing {name}")

with managed_resource("database") as res:
    print(f"  Using {res}")

# ===== Temporary Directory =====
@contextmanager
def temp_directory():
    """Create and clean up temp directory."""
    import tempfile
    tmpdir = tempfile.mkdtemp()
    print(f"  📁 Temp dir: {tmpdir}")
    try:
        yield Path(tmpdir)
    finally:
        import shutil
        shutil.rmtree(tmpdir)
        print(f"  🗑️ Removed: {tmpdir}")

with temp_directory() as tmp:
    (tmp / "test.txt").write_text("Hello!")

# ===== Change Working Directory =====
@contextmanager
def change_dir(new_dir: str):
    """Temporarily change working directory."""
    old_dir = os.getcwd()
    os.chdir(new_dir)
    try:
        yield
    finally:
        os.chdir(old_dir)

# ===== Redirect stdout =====
@contextmanager
def capture_output():
    """Capture stdout output."""
    from io import StringIO
    old_stdout = sys.stdout
    sys.stdout = buffer = StringIO()
    try:
        yield buffer
    finally:
        sys.stdout = old_stdout

with capture_output() as output:
    print("This is captured!")
    print("So is this!")

captured = output.getvalue()
print(f"Captured: {captured!r}")

# ===== Suppress and log errors =====
@contextmanager
def error_handler(raise_on_exit: bool = False):
    """Handle errors with logging."""
    errors = []
    try:
        yield errors
    except Exception as e:
        errors.append(str(e))
        print(f"  Logged error: {e}")
        if raise_on_exit:
            raise
    finally:
        if errors:
            print(f"  Total errors: {len(errors)}")

from contextlib import contextmanager
import threading
import time

# ===== Context Manager dengan setup dan teardown =====
@contextmanager
def database_transaction(connection):
    """Database transaction context manager."""
    cursor = connection.cursor()
    try:
        yield cursor
        connection.commit()
        print("  ✅ Transaction committed")
    except Exception as e:
        connection.rollback()
        print(f"  ❌ Transaction rolled back: {e}")
        raise
    finally:
        cursor.close()

# ===== Nested yield — multiple setup phases =====
@contextmanager
def phased_setup():
    """Context manager dengan beberapa fase setup."""
    print("  Phase 1: Initialize")
    data = {"initialized": True}

    try:
        print("  Phase 2: Configure")
        data["configured"] = True

        yield data  # Berikan data ke with block

        print("  Phase 5: Final cleanup")
    except Exception as e:
        print(f"  Error during phase: {e}")
        raise
    finally:
        print("  Phase 6: Always cleanup")

with phased_setup() as d:
    print(f"  Phase 3: Using data {d}")
    print("  Phase 4: Still using")

# ===== Thread-safe resource pool =====
@contextmanager
def resource_pool(resources: list):
    """Acquire resource from pool, release when done."""
    lock = threading.Lock()

    @contextmanager
    def acquire():
        with lock:
            if not resources:
                raise RuntimeError("No resources available")
            resource = resources.pop()
        try:
            yield resource
        finally:
            with lock:
                resources.append(resource)

    yield acquire

# Usage
pool = resource_pool(["conn1", "conn2", "conn3"])

with pool as get_resource:
    with get_resource() as conn:
        print(f"  Using {conn}")

# ===== Temporary environment variable =====
@contextmanager
def env_var(key: str, value: str):
    """Temporarily set environment variable."""
    old_value = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if old_value is None:
            del os.environ[key]
        else:
            os.environ[key] = old_value

import os
with env_var("DATABASE_URL", "sqlite:///:memory:"):
    print(f"  DB URL: {os.environ['DATABASE_URL']}")
print(f"  After: {os.environ.get('DATABASE_URL', 'not set')}")

from contextlib import (
    suppress,
    redirect_stdout,
    redirect_stderr,
    nullcontext,
    chdir,
    closing,
)
from pathlib import Path
from io import StringIO
import os

# ===== suppress: Menekan exception tertentu =====
with suppress(FileNotFoundError):
    Path("nonexistent.txt").unlink()
print("  Lanjut tanpa error!")

with suppress(FileNotFoundError, PermissionError):
    os.remove("/protected/file.txt")

# ===== redirect_stdout: Mengalihkan stdout =====
buffer = StringIO()
with redirect_stdout(buffer):
    print("Ini masuk ke buffer")
    print("Bukan ke terminal")

captured = buffer.getvalue()
print(f"  Captured: {captured!r}")

# Berguna untuk menangkap output library
with redirect_stdout(buffer) as buf:
    help(len)  # Output help() ke buffer

# ===== redirect_stderr: Mengalihkan stderr =====
err_buffer = StringIO()
with redirect_stderr(err_buffer):
    import sys
    print("Error message", file=sys.stderr)

print(f"  Stderr: {err_buffer.getvalue()!r}")

# ===== nullcontext: Placeholder CM =====
# Berguna ketika context manager opsional
def process(use_db: bool = False):
    # Jika use_db, gunakan DB connection; jika tidak, dummy
    cm = DatabaseConnection("app.db") if use_db else nullcontext()
    with cm as conn:
        print(f"  Processing with: {conn}")

# nullcontext dengan default value
cm = nullcontext(default_value="default_data")
with cm as value:
    print(f"  Value: {value}")  # "default_data"

# ===== closing: Auto-close objects =====
from urllib.request import urlopen

# closing() memanggil .close() pada objek saat keluar
with closing(urlopen("https://example.com")) as page:
    content = page.read()
# page.close() otomatis dipanggil

# ===== chdir (Python 3.11+) =====
# with chdir("/tmp"):
#     print(f"CWD: {os.getcwd()}")
# print(f"Back to: {os.getcwd()}")

from contextlib import contextmanager
import threading
import time

# ===== Nested with statements =====
@contextmanager
def connection(name: str):
    print(f"  🔌 Connect: {name}")
    try:
        yield name
    finally:
        print(f"  🔌 Disconnect: {name}")

@contextmanager
def transaction(conn_name: str):
    print(f"  💾 Begin transaction on {conn_name}")
    try:
        yield
        print(f"  ✅ Commit {conn_name}")
    except:
        print(f"  ❌ Rollback {conn_name}")
        raise

# Nested context managers
with connection("main_db") as conn:
    with transaction(conn):
        print(f"  📝 Executing queries on {conn}")

# ===== Multiple as (Python 3.1+) =====
with connection("db1") as c1, connection("db2") as c2:
    print(f"  Using {c1} and {c2}")

# ===== ExitStack — Dynamic nesting =====
from contextlib import ExitStack

def open_multiple_files(filenames: list):
    """Buka beberapa file sekaligus dengan ExitStack."""
    with ExitStack() as stack:
        files = [
            stack.enter_context(open(fname, "r"))
            for fname in filenames
        ]
        # Semua file terbuka
        for f in files:
            print(f"  {f.name}: {f.readline().strip()}")
    # Semua file ditutup

# ExitStack dengan callback cleanup
with ExitStack() as stack:
    # Register cleanup callbacks
    stack.callback(print, "  Cleanup 1")
    stack.callback(print, "  Cleanup 2")
    stack.callback(print, "  Cleanup 3")

    print("  Doing work...")
# Cleanup dipanggil dalam urutan terbalik (LIFO)

# Conditional context manager
with ExitStack() as stack:
    if True:
        f = stack.enter_context(open("output.txt", "w"))
    if False:  # Tidak masuk
        f2 = stack.enter_context(open("log.txt", "w"))

    print("  Done")

from contextlib import contextmanager, ContextDecorator
import threading

# ===== Reentrant CM: bisa digunakan berkali-kali =====
class IndentManager:
    """Reentrant context manager untuk indentation."""

    _level = 0

    def __enter__(self):
        IndentManager._level += 1
        return self

    def __exit__(self, *args):
        IndentManager._level -= 1
        return False

    @classmethod
    def print(cls, msg: str):
        print(f"{'  ' * cls._level}{msg}")

indent = IndentManager()
with indent:
    IndentManager.print("Level 1")
    with indent:
        IndentManager.print("Level 2")
        with indent:
            IndentManager.print("Level 3")
    IndentManager.print("Back to Level 1")

# ===== ContextDecorator: CM yang bisa jadi decorator =====
class LogEntryExit(ContextDecorator):
    """Context manager yang juga bisa dipakai sebagai decorator."""

    def __init__(self, name: str):
        self.name = name

    def __enter__(self):
        print(f"  → Enter: {self.name}")
        return self

    def __exit__(self, *args):
        print(f"  ← Exit: {self.name}")
        return False

# Sebagai context manager
with LogEntryExit("block"):
    print(f"  Doing work")

# Sebagai decorator
@LogEntryExit("my_function")
def my_function():
    print(f"  Function executing")

my_function()

# @contextmanager juga bisa jadi decorator
@contextmanager
def logged(func_name: str):
    print(f"  → {func_name}")
    try:
        yield
    finally:
        print(f"  ← {func_name}")

@logged("process_data")
def process_data():
    print(f"  Processing...")

process_data()

import asyncio
from contextlib import asynccontextmanager

# ===== Class-based Async CM =====
class AsyncDatabaseConnection:
    """Async context manager untuk database."""

    def __init__(self, db_url: str):
        self.db_url = db_url
        self.connection = None

    async def __aenter__(self):
        print(f"  🔌 Async connecting to {self.db_url}")
        await asyncio.sleep(0.5)  # Simulasi async connect
        self.connection = {"url": self.db_url, "connected": True}
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f"  ❌ Error: {exc_val}")
            await self._rollback()
        else:
            print(f"  ✅ Committing...")
            await self._commit()

        print(f"  🔒 Closing connection")
        await asyncio.sleep(0.2)  # Simulasi async close
        return False

    async def _commit(self):
        await asyncio.sleep(0.1)
        print("  💾 Committed")

    async def _rollback(self):
        await asyncio.sleep(0.1)
        print("  ↩️ Rolled back")

    async def execute(self, query: str):
        await asyncio.sleep(0.1)
        print(f"  📝 Executed: {query}")
        return [{"id": 1, "name": "Alice"}]

# ===== @asynccontextmanager =====
@asynccontextmanager
async def async_timer(label: str):
    """Async timer context manager."""
    start = asyncio.get_event_loop().time()
    yield
    elapsed = asyncio.get_event_loop().time() - start
    print(f"  ⏱️ {label}: {elapsed:.3f}s")

@asynccontextmanager
async def async_connection_pool(size: int = 5):
    """Async connection pool."""
    pool = [f"conn_{i}" for i in range(size)]
    print(f"  📦 Pool created with {size} connections")
    try:
        yield pool
    finally:
        print(f"  🗑️ Pool destroyed")

# ===== Main async function =====
async def main():
    # Async with class-based CM
    async with AsyncDatabaseConnection("sqlite:///app.db") as db:
        result = await db.execute("SELECT * FROM users")
        print(f"  Result: {result}")

    print()

    # Async with generator-based CM
    async with async_timer("Async Processing"):
        await asyncio.sleep(1.0)
        data = [x ** 2 for x in range(1000)]

    print()

    # Async pool
    async with async_connection_pool(3) as pool:
        conn = pool.pop()
        print(f"  Using {conn}")
        await asyncio.sleep(0.5)
        pool.append(conn)

asyncio.run(main())

import asyncio
from contextlib import AsyncExitStack, asynccontextmanager

@asynccontextmanager
async def managed_connection(name: str):
    print(f"  🔌 Open: {name}")
    await asyncio.sleep(0.2)
    try:
        yield name
    finally:
        print(f"  🔒 Close: {name}")

async def main():
    # Dynamic async context managers
    async with AsyncExitStack() as stack:
        connections = []
        for name in ["db", "cache", "queue"]:
            conn = await stack.enter_async_context(
                managed_connection(name)
            )
            connections.append(conn)

        print(f"  Active: {connections}")

        # Add cleanup callback
        stack.callback(lambda: print("  🧹 Final cleanup"))

    # All connections closed here

    # Async suppress
    from contextlib import asynccontextmanager

    @asynccontextmanager
    async def suppress_errors(*exceptions):
        try:
            yield
        except exceptions:
            pass

    async with suppress_errors(ValueError, TypeError):
        int("not_a_number")

asyncio.run(main())

from contextlib import contextmanager
import time
import random
from typing import Generator

@contextmanager
def retry_on_failure(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple = (Exception,)
) -> Generator[dict, None, None]:
    """Context manager dengan auto-retry dan exponential backoff."""
    state = {"attempt": 0, "last_error": None}

    class RetryProxy:
        def __init__(self):
            self.attempt = 0

        def should_retry(self) -> bool:
            return self.attempt < max_attempts

    proxy = RetryProxy()

    while proxy.should_retry():
        proxy.attempt += 1
        try:
            yield proxy
            return  # Success, exit
        except exceptions as e:
            proxy.last_error = e
            wait_time = delay * (backoff ** (proxy.attempt - 1))
            print(f"  ⚠️ Attempt {proxy.attempt} failed: {e}")
            if proxy.should_retry():
                print(f"  ⏳ Retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                print(f"  ❌ All {max_attempts} attempts failed!")
                raise

# Simulasi API yang kadang gagal
def unreliable_api():
    if random.random() < 0.7:  # 70% chance gagal
        raise ConnectionError("API timeout!")
    return {"status": "ok", "data": [1, 2, 3]}

# Menggunakan retry context manager
with retry_on_failure(max_attempts=5, delay=0.5) as retry:
    result = unreliable_api()
    print(f"  ✅ Success: {result}")

from contextlib import contextmanager
import json
from pathlib import Path
from typing import Dict, Any

class ConfigContextManager:
    """Context manager untuk aplikasi configuration."""

    def __init__(self, config_path: str = "config.json"):
        self.config_path = Path(config_path)
        self.config: Dict[str, Any] = {}
        self._defaults: Dict[str, Any] = {
            "debug": False,
            "log_level": "INFO",
            "database": {"host": "localhost", "port": 5432},
            "cache": {"enabled": True, "ttl": 300},
        }

    def __enter__(self):
        # Load config
        if self.config_path.exists():
            self.config = json.loads(self.config_path.read_text())
            print(f"  📖 Config loaded: {self.config_path}")
        else:
            self.config = self._defaults.copy()
            print(f"  📝 Using defaults")

        # Merge with defaults
        for key, value in self._defaults.items():
            if key not in self.config:
                self.config[key] = value

        return self.config

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            # Save config on successful exit
            self.config_path.write_text(
                json.dumps(self.config, indent=2)
            )
            print(f"  💾 Config saved: {self.config_path}")
        else:
            print(f"  ❌ Config NOT saved due to error")
        return False

    def get(self, key: str, default=None):
        return self.config.get(key, default)

# Usage
with ConfigContextManager("app_config.json") as config:
    config["debug"] = True
    config["database"]["host"] = "db.example.com"
    print(f"  Debug: {config['debug']}")
    print(f"  DB Host: {config['database']['host']}")
# Config otomatis disimpan

# ✅ DO: Selalu gunakan 'with' untuk file operations
with open("file.txt") as f:
    data = f.read()

# ❌ DON'T: Buka file tanpa with
f = open("file.txt")
data = f.read()
f.close()

# ✅ DO: Gunakan @contextmanager untuk CM sederhana
from contextlib import contextmanager

@contextmanager
def managed_resource():
    resource = acquire()
    try:
        yield resource
    finally:
        release(resource)

# ✅ DO: Gunakan class-based untuk CM kompleks
class ComplexResource:
    def __init__(self, config):
        self.config = config

    def __enter__(self):
        self._setup()
        return self

    def __exit__(self, *exc_info):
        self._cleanup()
        return False

    def _setup(self):
        pass

    def _cleanup(self):
        pass

# ✅ DO: Selalu re-raise exception di @contextmanager
@contextmanager
def safe_cm():
    try:
        yield
    except Exception:
        # Log error
        raise  # Re-raise!
    finally:
        cleanup()

# ❌ DON'T: Swallow exception tanpa re-raise
@contextmanager
def bad_cm():
    try:
        yield
    except Exception:
        pass  # Jangan lakukan ini!

# ✅ DO: Gunakan nullcontext() untuk optional CM
def process(use_lock=False):
    lock = threading.Lock() if use_lock else nullcontext()
    with lock:
        do_work()

# ✅ DO: Kembalikan self dari __enter__ jika memungkinkan
def __enter__(self):
    return self  # Agar bisa: with cm as instance: