Какие знаешь ключевые слова в контекстном менеджере?

Ключевые слова в контекстном менеджере

Контекстный менеджер в Python — это объект, который управляет ресурсами. Основные ключевые слова: with, as, __enter__, __exit__.

Ключевое слово: with

# Открывает контекстный менеджер
with open('file.txt') as f:
    data = f.read()
# Автоматически закрывает файл

# Эквивалент (без with)
f = open('file.txt')
try:
    data = f.read()
finally:
    f.close()  # Гарантированно выполнится

Ключевое слово: as

# Привязывает результат __enter__() к переменной
with open('file.txt') as f:  # 'f' = результат __enter__()
    content = f.read()

# Можно без as
with open('file.txt'):
    pass

# Множественные менеджеры
with open('input.txt') as f_in, open('output.txt', 'w') as f_out:
    f_out.write(f_in.read())

Магические методы: __enter__ и __exit__

class MyContext:
    def __enter__(self):
        print("Entering")
        return self  # Это привязывается к переменной после 'as'
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Exiting")
        # exc_type: тип исключения (None если нет)
        # exc_val: значение исключения
        # exc_tb: traceback
        return False  # Пробросить исключение дальше

with MyContext() as ctx:
    print("Inside")
# Вывод:
# Entering
# Inside
# Exiting

Примеры контекстных менеджеров

1. Файлы:

with open('data.txt') as f:
    lines = f.readlines()

# __exit__ закроет файл автоматически

2. Блокировки (Threading):

import threading

lock = threading.Lock()

with lock:
    # Потокобезопасная операция
    shared_resource.modify()
# Lock автоматически отпущен

3. Контроль времени:

import time
from contextlib import contextmanager

@contextmanager
def timer():
    start = time.time()
    yield
    elapsed = time.time() - start
    print(f"Elapsed: {elapsed:.2f}s")

with timer():
    time.sleep(1)
# Elapsed: 1.00s

4. Подавление исключений:

from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove('file.txt')
# Если файл не существует, ошибка будет проигнорирована

5. Временное изменение состояния:

from contextlib import contextmanager

@contextmanager
def change_cwd(path):
    old_cwd = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old_cwd)

with change_cwd('/tmp'):
    print(os.getcwd())  # /tmp
print(os.getcwd())  # back to original

Обработка исключений в exit

class ErrorHandler:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            print(f"Caught ValueError: {exc_val}")
            return True  # Подавить исключение
        elif exc_type is KeyError:
            print(f"Caught KeyError: {exc_val}")
            return True  # Подавить
        return False  # Пробросить дальше

with ErrorHandler():
    raise ValueError("Something bad")  # Будет подавлено

with ErrorHandler():
    raise RuntimeError("Something worse")  # Будет проброшено

@contextmanager декоратор

from contextlib import contextmanager

@contextmanager
def my_context():
    print("Enter")
    try:
        yield "resource"
    finally:
        print("Exit")

with my_context() as resource:
    print(f"Using {resource}")

# Вывод:
# Enter
# Using resource
# Exit

AsyncIO контекстные менеджеры

from aiofiles import open as aopen

async def read_file():
    async with aopen('file.txt') as f:
        content = await f.read()
    return content

# Магические методы: __aenter__ и __aexit__
class AsyncContext:
    async def __aenter__(self):
        print("Entering async")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Exiting async")
        return False

async def main():
    async with AsyncContext():
        print("Inside")

Практические примеры

1. Connection pooling:

from contextlib import contextmanager

class Database:
    @contextmanager
    def connection(self):
        conn = self.pool.acquire()
        try:
            yield conn
        finally:
            self.pool.release(conn)

db = Database()
with db.connection() as conn:
    result = conn.execute("SELECT * FROM users")

2. Транзакции:

from contextlib import contextmanager

@contextmanager
def transaction(db):
    db.begin()
    try:
        yield
        db.commit()
    except:
        db.rollback()
        raise

with transaction(db):
    db.insert('users', {'name': 'John'})
    db.insert('users', {'name': 'Jane'})
# Оба инсерта в одной транзакции

3. Временное переопределение:

from unittest.mock import patch

with patch('requests.get') as mock_get:
    mock_get.return_value.status_code = 200
    response = requests.get('https://api.example.com')
    assert response.status_code == 200
# requests.get вернул в нормальное состояние

Преимущества контекстных менеджеров

# Плохо: легко забыть close()
f = open('file.txt')
data = f.read()
# Забыли f.close()!

# Хорошо: гарантированное закрытие
with open('file.txt') as f:
    data = f.read()
# Закроется автоматически, даже если исключение

Вывод: Контекстные менеджеры — это мощный инструмент для управления ресурсами. Ключевые слова with и as упрощают работу с файлами, блокировками и другими ресурсами, гарантируя правильную очистку.