Python Context Managers

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')}")