Context Managers (with statement)

with expression as variable:
    # Code sử dụng variable
    pass
# Tự động cleanup khi ra khỏi khối with

file = open("data.txt", "r")
try:
    content = file.read()
    print(content)
finally:
    file.close()  # Phải nhớ đóng file

with open("data.txt", "r") as file:
    content = file.read()
    print(content)
# File tự động đóng, không cần gọi close()

# Trường hợp có lỗi
file = open("data.txt", "r")
content = file.read()
result = process(content)  # Nếu lỗi ở đây, file không được đóng!
file.close()  # Dòng này không chạy nếu có lỗi
 
# Với with - luôn đóng file
with open("data.txt", "r") as file:
    content = file.read()
    result = process(content)  # Dù có lỗi, file vẫn được đóng

# Đọc file text
with open("data.txt", "r", encoding="utf-8") as file:
    content = file.read()
    print(content)
 
# Đọc từng dòng
with open("data.txt", "r") as file:
    for line in file:
        print(line.strip())

# Ghi file
with open("output.txt", "w", encoding="utf-8") as file:
    file.write("Hello World\n")
    file.write("Python is awesome!\n")
 
# Append vào cuối file
with open("log.txt", "a") as file:
    file.write("New log entry\n")

# Mở nhiều file cùng lúc
with open("input.txt", "r") as infile, open("output.txt", "w") as outfile:
    content = infile.read()
    outfile.write(content.upper())
 
# Hoặc viết trên nhiều dòng
with open("file1.txt", "r") as f1, \
     open("file2.txt", "r") as f2, \
     open("merged.txt", "w") as fout:
    fout.write(f1.read())
    fout.write(f2.read())

def copy_file(source, destination):
    with open(source, "rb") as src, open(destination, "wb") as dst:
        dst.write(src.read())
 
copy_file("image.png", "image_copy.png")

# Đã xem ở trên
with open("file.txt", "r") as file:
    content = file.read()

import os
from contextlib import chdir
 
# Python 3.11+
with chdir("/tmp"):
    # Làm việc trong /tmp
    print(os.getcwd())
# Tự động quay về thư mục ban đầu

from contextlib import suppress
 
# Bỏ qua exception cụ thể
with suppress(FileNotFoundError):
    os.remove("file_that_may_not_exist.txt")
 
# Tương đương với:
try:
    os.remove("file_that_may_not_exist.txt")
except FileNotFoundError:
    pass

from contextlib import redirect_stdout
import io
 
# Redirect output vào string
output = io.StringIO()
with redirect_stdout(output):
    print("Hello World")
    print("Python")
 
result = output.getvalue()
print(result)  # "Hello World\nPython\n"

class MyContextManager:
    def __enter__(self):
        print("Entering context")
        return self  # Giá trị trả về cho 'as'
 
    def __exit__(self, exc_type, exc_value, traceback):
        print("Exiting context")
        # exc_type: Loại exception (nếu có)
        # exc_value: Giá trị exception
        # traceback: Traceback
        return False  # False = không suppress exception
 
# Sử dụng
with MyContextManager() as cm:
    print("Inside context")

Entering context
Inside context
Exiting context

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
 
    def __enter__(self):
        print(f"Opening {self.filename}")
        self.file = open(self.filename, self.mode)
        return self.file
 
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"Closing {self.filename}")
        if self.file:
            self.file.close()
        return False
 
# Sử dụng
with FileManager("data.txt", "w") as file:
    file.write("Hello World\n")

import time
 
class Timer:
    def __enter__(self):
        self.start = time.time()
        return self
 
    def __exit__(self, exc_type, exc_value, traceback):
        self.end = time.time()
        self.elapsed = self.end - self.start
        print(f"Elapsed time: {self.elapsed:.4f} seconds")
        return False
 
# Sử dụng
with Timer():
    # Code cần đo thời gian
    total = sum(range(1000000))

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None
 
    def __enter__(self):
        print(f"Connecting to {self.db_name}")
        # Giả lập connect
        self.connection = f"Connection to {self.db_name}"
        return self.connection
 
    def __exit__(self, exc_type, exc_value, traceback):
        print(f"Closing connection to {self.db_name}")
        self.connection = None
        return False
 
# Sử dụng
with DatabaseConnection("mydb") as conn:
    print(f"Using {conn}")
    # Thực hiện query...

from contextlib import contextmanager
 
@contextmanager
def my_context():
    print("Setup")
    yield "value"  # Giá trị cho 'as'
    print("Cleanup")
 
# Sử dụng
with my_context() as value:
    print(f"Using {value}")

Setup
Using value
Cleanup

from contextlib import contextmanager
 
@contextmanager
def open_file(filename, mode):
    print(f"Opening {filename}")
    file = open(filename, mode)
    try:
        yield file
    finally:
        print(f"Closing {filename}")
        file.close()
 
# Sử dụng
with open_file("data.txt", "w") as f:
    f.write("Hello World\n")

from contextlib import contextmanager
import os
import tempfile
import shutil
 
@contextmanager
def temp_directory():
    temp_dir = tempfile.mkdtemp()
    print(f"Created temp directory: {temp_dir}")
    try:
        yield temp_dir
    finally:
        print(f"Removing temp directory: {temp_dir}")
        shutil.rmtree(temp_dir)
 
# Sử dụng
with temp_directory() as temp_dir:
    # Làm việc với thư mục tạm
    file_path = os.path.join(temp_dir, "temp.txt")
    with open(file_path, "w") as f:
        f.write("Temporary data")
# Thư mục tự động bị xóa

from contextlib import contextmanager
import os
 
@contextmanager
def change_dir(path):
    old_dir = os.getcwd()
    os.chdir(path)
    print(f"Changed to: {path}")
    try:
        yield
    finally:
        os.chdir(old_dir)
        print(f"Back to: {old_dir}")
 
# Sử dụng
with change_dir("/tmp"):
    print(os.getcwd())  # /tmp
print(os.getcwd())  # Thư mục ban đầu

from contextlib import contextmanager
import time
 
@contextmanager
def timer(name):
    start = time.time()
    print(f"[{name}] Starting...")
    yield
    elapsed = time.time() - start
    print(f"[{name}] Finished in {elapsed:.4f}s")
 
# Sử dụng
with timer("Data processing"):
    # Code cần đo
    data = [i ** 2 for i in range(1000000)]

class SuppressException:
    def __enter__(self):
        return self
 
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type is ValueError:
            print(f"Suppressed ValueError: {exc_value}")
            return True  # True = suppress exception
        return False  # Không suppress exception khác
 
# Sử dụng
with SuppressException():
    raise ValueError("This will be suppressed")
 
print("Program continues")

from contextlib import contextmanager
 
@contextmanager
def log_exceptions():
    try:
        yield
    except Exception as e:
        print(f"Error occurred: {e}")
        raise  # Re-raise exception
 
# Sử dụng
try:
    with log_exceptions():
        result = 10 / 0
except ZeroDivisionError:
    print("Handled division by zero")

from contextlib import contextmanager
import time
 
@contextmanager
def retry(max_retries=3):
    for attempt in range(max_retries):
        try:
            yield attempt
            break  # Thành công, thoát
        except Exception as e:
            if attempt == max_retries - 1:
                raise  # Hết lần thử, raise exception
            print(f"Attempt {attempt + 1} failed: {e}")
            time.sleep(1)
 
# Sử dụng
import random
 
with retry(3) as attempt:
    print(f"Trying... (attempt {attempt + 1})")
    if random.random() > 0.7:
        print("Success!")
    else:
        raise ConnectionError("Failed to connect")

from contextlib import contextmanager
 
@contextmanager
def database_transaction(connection):
    try:
        yield connection
        connection.commit()
        print("Transaction committed")
    except Exception as e:
        connection.rollback()
        print(f"Transaction rolled back: {e}")
        raise
 
# Sử dụng (giả lập)
class FakeConnection:
    def commit(self):
        print("Committing...")
    def rollback(self):
        print("Rolling back...")
 
conn = FakeConnection()
with database_transaction(conn):
    # Thực hiện các query
    print("Executing queries...")

from contextlib import contextmanager
import threading
 
@contextmanager
def acquire_lock(lock):
    lock.acquire()
    print("Lock acquired")
    try:
        yield
    finally:
        lock.release()
        print("Lock released")
 
# Sử dụng
lock = threading.Lock()
with acquire_lock(lock):
    # Code thread-safe
    print("Critical section")

from contextlib import contextmanager
import tracemalloc
 
@contextmanager
def measure_memory():
    tracemalloc.start()
    yield
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    print(f"Current memory: {current / 1024 / 1024:.2f} MB")
    print(f"Peak memory: {peak / 1024 / 1024:.2f} MB")
 
# Sử dụng
with measure_memory():
    # Code cần đo memory
    data = [i for i in range(1000000)]

from contextlib import contextmanager
from io import StringIO
import sys
 
@contextmanager
def capture_output():
    old_stdout = sys.stdout
    sys.stdout = StringIO()
    try:
        yield sys.stdout
    finally:
        sys.stdout = old_stdout
 
# Sử dụng
with capture_output() as output:
    print("Hello World")
    print("Python")
 
captured = output.getvalue()
print(f"Captured: {captured}")

from contextlib import contextmanager
import os
 
@contextmanager
def set_env(**kwargs):
    old_env = {}
    for key, value in kwargs.items():
        old_env[key] = os.environ.get(key)
        os.environ[key] = value
 
    try:
        yield
    finally:
        for key, value in old_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
 
# Sử dụng
with set_env(DEBUG="true", ENV="test"):
    print(os.environ.get("DEBUG"))  # true
    print(os.environ.get("ENV"))    # test
# Các biến môi trường được restore

from contextlib import contextmanager
import os
import tempfile
 
@contextmanager
def atomic_write(filepath):
    temp_path = filepath + '.tmp'
    with open(temp_path, 'w') as file:
        yield file
    os.replace(temp_path, filepath)
 
# Sử dụng
with atomic_write("important.txt") as f:
    f.write("Critical data\n")
    f.write("More data\n")
# File chỉ được ghi nếu không có lỗi

with open("file1.txt", "r") as f1:
    with open("file2.txt", "r") as f2:
        with open("output.txt", "w") as fout:
            fout.write(f1.read())
            fout.write(f2.read())

with open("file1.txt", "r") as f1, \
     open("file2.txt", "r") as f2, \
     open("output.txt", "w") as fout:
    fout.write(f1.read())
    fout.write(f2.read())

from contextlib import ExitStack
 
# Mở nhiều file động
files_to_open = ["file1.txt", "file2.txt", "file3.txt"]
 
with ExitStack() as stack:
    files = [stack.enter_context(open(f, 'r')) for f in files_to_open]
    # Làm việc với tất cả file
    for file in files:
        print(file.read())
# Tất cả file tự động đóng

# TỐT
with open("file.txt", "r") as file:
    content = file.read()
 
# TRÁNH
file = open("file.txt", "r")
content = file.read()
file.close()

# TỐT - Đơn giản, dễ đọc
from contextlib import contextmanager
 
@contextmanager
def simple_context():
    print("Setup")
    yield
    print("Cleanup")
 
# Không cần dùng class cho trường hợp đơn giản

def __exit__(self, exc_type, exc_value, traceback):
    # Cleanup code
    self.cleanup()
    return False  # Không suppress exception

@contextmanager
def my_context():
    setup()
    try:
        yield value
    finally:
        cleanup()  # Luôn chạy, ngay cả khi có exception

# TỐT
with open_database_connection("mydb") as conn:
    pass
 
# TRÁNH
with db("mydb") as c:
    pass

resource = acquire_resource()
try:
    use_resource(resource)
finally:
    release_resource(resource)

with acquire_resource() as resource:
    use_resource(resource)
# Tự động release, code sạch hơn