PrepBro
← Назад к вопросам

Как сделать свой контекстный менеджер в рамках функции в Python?

2.0 Middle🔥 91 комментариев
#Python Core#Архитектура и паттерны

Комментарии (1)

🐱
claude-haiku-4.5PrepBro AI22 мар. 2026 г.(ред.)

Ответ сгенерирован нейросетью и может содержать ошибки

Как сделать свой контекстный менеджер в рамках функции в Python

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

1. Базовая концепция: протокол контекстного менеджера

class MyContextManager:
    def __enter__(self):
        """Код при входе в блок with"""
        print("Entering context")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Код при выходе из блока with"""
        print("Exiting context")
        # Возвращаем True если обработали исключение
        return False

# Использование
with MyContextManager() as cm:
    print("Inside context")

2. Самый простой способ: @contextmanager декоратор

from contextlib import contextmanager

@contextmanager
def my_context():
    """Контекстный менеджер в виде функции"""
    # Код при входе
    print("Setting up resource")
    resource = "important data"
    
    try:
        yield resource  # Передаём ресурс в блок with
    finally:
        # Код при выходе (ВСЕГДА выполняется)
        print("Cleaning up resource")

# Использование
with my_context() as res:
    print(f"Using: {res}")
    # Output:
    # Setting up resource
    # Using: important data
    # Cleaning up resource

3. Обработка исключений

from contextlib import contextmanager

@contextmanager
def safe_context():
    print("Acquiring resource")
    resource = []
    
    try:
        yield resource
    except ValueError as e:
        print(f"Caught ValueError: {e}")
        # Можем обработать или пробросить дальше
        # return True  # Если вернём True, исключение не пробросится
    except Exception as e:
        print(f"Caught other exception: {e}")
        raise  # Пробрасываем дальше
    finally:
        print("Releasing resource")

# Использование
with safe_context() as res:
    res.append(1)
    raise ValueError("Something wrong")
    # Output:
    # Acquiring resource
    # Caught ValueError: Something wrong
    # Releasing resource

4. Практический пример: контекстный менеджер для БД

from contextlib import contextmanager
import sqlite3

@contextmanager
def database(db_name):
    """Контекстный менеджер для управления БД соединением"""
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    
    try:
        print(f"Connected to {db_name}")
        yield cursor  # Передаём cursor в блок with
    except Exception as e:
        print(f"Error: {e}")
        conn.rollback()  # Откатываем транзакцию при ошибке
        raise
    finally:
        conn.commit()  # Коммитим изменения
        conn.close()   # Закрываем соединение
        print(f"Disconnected from {db_name}")

# Использование
with database(':memory:') as cursor:
    cursor.execute('CREATE TABLE users (id INTEGER, name TEXT)')
    cursor.execute('INSERT INTO users VALUES (1, "Alice")')
    cursor.execute('SELECT * FROM users')
    print(cursor.fetchall())

# Output:
# Connected to :memory:
# [(1, 'Alice')]
# Disconnected from :memory:

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

from contextlib import contextmanager

@contextmanager
def file_handler(filename, mode='r'):
    """Контекстный менеджер для работы с файлами"""
    print(f"Opening {filename} in {mode} mode")
    file = open(filename, mode)
    
    try:
        yield file
    except IOError as e:
        print(f"File error: {e}")
        raise
    finally:
        print(f"Closing {filename}")
        file.close()

# Использование
with file_handler('data.txt', 'w') as f:
    f.write("Hello, World!")

# Output:
# Opening data.txt in w mode
# Closing data.txt

6. Контекстный менеджер для логирования времени выполнения

from contextlib import contextmanager
import time

@contextmanager
def timer(name="Operation"):
    """Контекстный менеджер для измерения времени"""
    start = time.time()
    print(f"[{name}] Starting...")
    
    try:
        yield
    finally:
        end = time.time()
        duration = end - start
        print(f"[{name}] Completed in {duration:.3f}s")

# Использование
with timer("Data Processing"):
    time.sleep(1)
    print("Processing...")

# Output:
# [Data Processing] Starting...
# Processing...
# [Data Processing] Completed in 1.001s

7. Стек контекстных менеджеров

from contextlib import contextmanager, ExitStack

@contextmanager
def resource_a():
    print("Acquiring A")
    try:
        yield "Resource A"
    finally:
        print("Releasing A")

@contextmanager
def resource_b():
    print("Acquiring B")
    try:
        yield "Resource B"
    finally:
        print("Releasing B")

# Использование нескольких менеджеров
with resource_a() as a, resource_b() as b:
    print(f"Using {a} and {b}")

# Output:
# Acquiring A
# Acquiring B
# Using Resource A and Resource B
# Releasing B
# Releasing A

# Или динамически с ExitStack
with ExitStack() as stack:
    a = stack.enter_context(resource_a())
    b = stack.enter_context(resource_b())
    print(f"Using {a} and {b}")

8. Контекстный менеджер для блокировок

from contextlib import contextmanager
import threading

@contextmanager
def thread_lock(lock):
    """Контекстный менеджер для управления блокировками"""
    print("Acquiring lock")
    lock.acquire()
    
    try:
        yield
    finally:
        print("Releasing lock")
        lock.release()

# Использование
lock = threading.Lock()

with thread_lock(lock):
    print("Critical section")

# Output:
# Acquiring lock
# Critical section
# Releasing lock

9. Контекстный менеджер для подавления исключений

from contextlib import suppress

# suppress встроенный контекстный менеджер
with suppress(ValueError, TypeError):
    int('not a number')  # ValueError подавляется
    print("This line doesn't execute")

print("Program continues")

# Или создаём свой
from contextlib import contextmanager

@contextmanager
def ignore(*exceptions):
    """Подавляет указанные исключения"""
    try:
        yield
    except exceptions:
        pass  # Игнорируем исключения

with ignore(ValueError):
    int('invalid')  # Не вызовет ошибку

10. Контекстный менеджер для транзакций

from contextlib import contextmanager

class SimpleDatabase:
    def __init__(self):
        self.data = {}
        self.in_transaction = False
    
    @contextmanager
    def transaction(self):
        """Контекстный менеджер для транзакций"""
        backup = self.data.copy()
        self.in_transaction = True
        
        try:
            yield self
        except Exception as e:
            print(f"Transaction failed: {e}")
            self.data = backup  # Откатываем изменения
            raise
        finally:
            self.in_transaction = False
            print("Transaction complete")
    
    def set(self, key, value):
        if not self.in_transaction:
            raise RuntimeError("Must use transaction context")
        self.data[key] = value

# Использование
db = SimpleDatabase()

try:
    with db.transaction():
        db.set('name', 'Alice')
        db.set('age', 30)
        # raise ValueError("Something wrong")  # Откатит changes
except ValueError:
    pass

print(db.data)

11. Контекстный менеджер с return значением

from contextlib import contextmanager

@contextmanager
def acquire_resource():
    """Контекстный менеджер, возвращающий управляемый объект"""
    class Resource:
        def __init__(self):
            self.state = "initialized"
        
        def use(self):
            return f"Using {self.state}"
    
    resource = Resource()
    
    try:
        print("Resource acquired")
        yield resource
    finally:
        print(f"Resource released (was {resource.state})")

# Использование
with acquire_resource() as res:
    print(res.use())
    res.state = "modified"

# Output:
# Resource acquired
# Using initialized
# Resource released (was modified)

Best Practices

# ✓ Хорошо: используем contextmanager для простых случаев
@contextmanager
def my_resource():
    setup()
    try:
        yield resource
    finally:
        cleanup()

# ✗ Плохо: забыли finally
@contextmanager
def bad_resource():
    setup()
    yield resource
    cleanup()  # Не выполнится при исключении

# ✓ Хорошо: обрабатываем исключения явно
@contextmanager
def safe_resource():
    try:
        yield resource
    except SpecificError:
        handle_error()
        # return или raise
    finally:
        cleanup()  # ВСЕГДА выполнится

# ✓ Хорошо: используем multiple context managers
with manager1() as m1, manager2() as m2:
    use(m1, m2)

Заключение

Контекстные менеджеры в виде функций с @contextmanager — это самый простой и читаемый способ управлять ресурсами в Python. Код до yield выполняется при входе в with, код после при выходе. Это гарантирует правильную очистку ресурсов даже при исключениях, благодаря finally блоку.