Как работают контекстные операторы?

Как работают контекстные менеджеры в Python

Контекстные менеджеры (with statement) — один из самых элегантных паттернов Python для управления ресурсами.

Проблема: ручное управление ресурсами

# Плохо
f = open('file.txt')
data = f.read()
f.close()  # Что если будет исключение?

# Ещё хуже
f = open('file.txt')
data = f.read()
if 'error' in data:
    raise ValueError("Error found")
f.close()  # Никогда не выполнится!

Решение: контекстный менеджер

# Хорошо
with open('file.txt') as f:
    data = f.read()
    # Даже если исключение — файл закроется!

print(f.closed)  # True

Как работает enter и exit

class FileManager:
    def __init__(self, filename):
        self.filename = filename
        self.file = None
    
    def __enter__(self):
        print(f"Opening {self.filename}")
        self.file = open(self.filename)
        return self.file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"Closing {self.filename}")
        if self.file:
            self.file.close()
        return False

with FileManager('data.txt') as f:
    data = f.read()

Практический пример: управление БД подключением

import psycopg2

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.conn = None
    
    def __enter__(self):
        self.conn = psycopg2.connect(self.connection_string)
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.conn:
            if exc_type:
                self.conn.rollback()  # Откатываем при ошибке
            else:
                self.conn.commit()  # Коммитим при успехе
            self.conn.close()
        return False

with DatabaseConnection('postgresql://localhost/db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()

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

# Файлы
with open('file.txt') as f:
    data = f.read()

# Блокировки
import threading
lock = threading.Lock()
with lock:
    pass

# Подавление исключений
from contextlib import suppress
with suppress(FileNotFoundError):
    os.remove('nonexistent.txt')

Декоратор @contextmanager

from contextlib import contextmanager

@contextmanager
def database_connection(connection_string):
    conn = psycopg2.connect(connection_string)
    try:
        yield conn  # Передаём ресурс блоку with
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.commit()
        conn.close()

with database_connection('postgresql://localhost/db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')

Контекстный менеджер с параметрами

from contextlib import contextmanager
import time

@contextmanager
def time_logger(operation_name):
    start = time.time()
    print(f"Starting {operation_name}")
    try:
        yield
    finally:
        elapsed = time.time() - start
        print(f"Finished in {elapsed:.2f}s")

with time_logger("database query"):
    db.execute_query()

Множественные контекстные менеджеры

# Python 3.10+
with open('input.txt') as inp, open('output.txt', 'w') as out:
    for line in inp:
        out.write(line.upper())

# Или старый синтаксис
with open('input.txt') as inp:
    with open('output.txt', 'w') as out:
        for line in inp:
            out.write(line.upper())

Практический пример: Pytest фикстура

from contextlib import contextmanager

@contextmanager
def temp_database():
    db = create_temp_database()
    try:
        yield db
    finally:
        drop_temp_database(db)

def test_something():
    with temp_database() as db:
        user = db.create_user('Alice')
        assert user.name == 'Alice'
        # БД автоматически удалена

Что происходит при исключении

class ExceptionHandler:
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            print(f"Caught: {exc_type.__name__}")
        return False  # Пробросить исключение

with ExceptionHandler():
    raise ValueError("Test")  # Выведет сообщение

Практическое применение

• Файлы: автоматически закрыть
• БД: автоматически коммит/откат и закрытие
• Блокировки: автоматически разблокировать
• Временные ресурсы: автоматически очистить
• Логирование: логировать начало и конец операции
• Изоляция: создать и удалить временное окружение