Теория
Оператор with появился в python 2.5, но, не смотря на это, используется до сих пор недостаточно широко. Являясь упрощенной версией анонимных блоков кода with позволяет:
- исполнить код до начала блока
- исполнить код по выходу из блока, независимо от того это выход по исключению с помощью return или другим способом
- обработать исключение, возникшее в блоке.
Синтаксически with выглядит следующим образом:
with operation:
code
with operation:
code
operation может быть объектом, выражением или конструкцией вида expression as var. Как и много других конструкций он является синтаксическим сахаром для более громоздкого выражения:
with operation as var:
code
with operation as var:
code
=>
_obj = operation
# вход в блок
var = _obj.__enter__()
try:
code
except Exception as exc:
# если произошло исключение - передаем его управляющему объекту
if not _obj.__exit__(*sys.exception_info()):
# если он вернул False(None) возбуждаем его
raise
# если True - подавляем исключение
else:
# если не было исключения - передаем None * 3
_obj.__exit__(None, None, None)
_obj = operation
# вход в блок
var = _obj.__enter__()
try:
code
except Exception as exc:
# если произошло исключение - передаем его управляющему объекту
if not _obj.__exit__(*sys.exception_info()):
# если он вернул False(None) возбуждаем его
raise
# если True - подавляем исключение
else:
# если не было исключения - передаем None * 3
_obj.__exit__(None, None, None)
Более подробно с with можно ознакомиться в соответствующем PEP-343. with управляется объектом, называемым менеджером контекста (МК) — _obj в примере выше. Есть два основных способа написания МК — класс с методами __enter__ и __exit__ и генератор:
import os
from contextlib import contextmanager
# Это только пример.
# Использование такого кода для генерации временных файлов
# небезопасно. Используйте функции 'os.tmpfile'.
class TempoFileCreator(object):
def __init__(self):
self.fname = None
self.fd = None<
br />
def __inter__(self):
# вызывается по входу в блок
self.fname = os.tmpnam()
self.fd = open(self.fname, "w+")
return self.fname, self.fd
def __exit__(self, exc_type, exc_val, traceback):
# вызывается по выходу из блока
# если в блоке выброшено исключение, то
# его тип, значение и трейс будут переданы в параметрах
self.fd.close()
os.unlink(self.fname)
self.fd = None
self.fname = None
# здесь написано return None => исключение не будет подавляться
@contextmanager
def tempo_file():
# полностью равноценно классу TempoFileCreator
fname = os.tmpnam()
fd = open(fname, "w+")
try:
yield fname, fd
#сейчас исполняется блок
finally:
# это наш __exit__
fd.close()
os.unlink(fd)
import os
from contextlib import contextmanager
# Это только пример.
# Использование такого кода для генерации временных файлов
# небезопасно. Используйте функции 'os.tmpfile'.
class TempoFileCreator(object):
def __init__(self):
self.fname = None
self.fd = None
def __inter__(self):
# вызывается по входу в блок
self.fname = os.tmpnam()
self.fd = open(self.fname, "w+")
return self.fname, self.fd
def __exit__(self, exc_type, exc_val, traceback):
# вызывается по выходу из блока
# если в блоке выброшено исключение, то
# его тип, значение и трейс будут переданы в параметрах
self.fd.close()
os.unlink(self.fname)
self.fd = None
self.fname = None
# здесь написано return None => исключение не будет подавляться
@contextmanager
def tempo_file():
# полностью равноценно классу TempoFileCreator
fname = os.tmpnam()
fd = open(fname, "w+")
try:
yield fname, fd
#сейчас исполняется блок
finally:
# это наш __exit__
fd.close()
os.unlink(fd)
Использование:
with tempo_file() as (fname, fd):
# читаем-пишем в файл
# по выходу из блока он будет удален
pass
with tempo_file() as (fname, fd):
# читаем-пишем в файл
# по выходу из блока он будет удален
pass
Ядро python реализует только первый вариант для контекст менеджера, второй реализуется в contextlib.contextmanager.
В том случае если во внутреннем блоке кода есть оператор yield, т.е. мы работаем в генераторе, __exit__ будет вызван по выходу из генератора или по его удалению. Таким образом если ссылку на генератор сохранить, то __exit__ не будет вызван до тех пор, пока ссылка будет существовать:
@contextmanager
def cmanager():
yield
print "Exit"
def some_func():
with cmanager():
yield 1
it = some_func()
for val in it:
pass
# Exit напечатается здесь
it = some_func()
del it # или по выходу из текущего блока
# Exit напечатается здесь
@contextmanager
def cmanager():
yield
print "Exit"
def some_func():
with cmanager():
yield 1
it = some_func()
for val in it:
pass
# Exit напечатается здесь
it = some_func()
del it # или по выходу из текущего блока
# Exit напечатается здесь
Подводя итоги — with позволяет сэкономить 2-4 строки кода на каждое использование и повышает читаемость программы, меньше отвлекая нас от логики деталями реализации.
Практика
Начнем с примеров, которые встречаются в стандартной библиотеке и будем постепенно переходить к менее распространенным вариантам использования.
- Открытие/создание объекта по входу в блок — закрытие/удаление по выходу:
with open('/tmp/tt.txt') as fd:
pass
# здесь файл закрывается
# переменная fd доступна, но файл уже закрыт
#
with open('/tmp/tt.txt') as fd:
pass
# здесь файл закрывается
# переменная fd доступна, но файл уже закрыт
# Чаще всего в python программах не закрывают файл вручную, обоснованно полагаясь на подсчет ссылок. Блоки with кроме явного указания области, где файл открыт имеют еще одно небольшое преимущество, связанное с особенностями обработки исключений:
def i_am_not_always_close_files(fname):
fd = open(fname)
i_am_not_always_close_files("/tmp/x.txt")
# в этой точке файл уже закрыт
def i_am_not_always_close_files(fname):
fd = open(fname)
i_am_not_always_close_files("/tmp/x.txt")
# в этой точке файл уже закрыт
Если внутри фцнкции i_am_not_always_close_files будет возбуждено исключение, то файл не закроется до того момента, пока оно не будет обработано:
import sys
def i_am_not_always_close_files(fname):
fd = open(fname)
raise RuntimeError('')
try:
i_am_not_always_close_files("/tmp/x.txt")
except RuntimeError:
#тут файл еще открыт
traceback = sys.exc_info()[2]
# спуск на один кадр стека глубже
# 'fd' в его локальных переменных
print traceback.tb_next.tb_frame.f_locals['fd']
#
# в этой точке файл уже закрыт
import sys
def i_am_not_always_close_files(fname):
fd = open(fname)
raise RuntimeError('')
try:
i_am_not_always_close_files("/tmp/x.txt")
except RuntimeError:
#тут файл еще открыт
traceback = sys.exc_info()[2]
# спуск на один кадр стека глубже
# 'fd' в его локальных переменных
print traceback.tb_next.tb_frame.f_locals['fd']
#
# в этой точке файл уже закрыт
Дело в том, что пока жив объект-исключение он хранит путь исключения со всеми кадрами стека. При использовании with файл закрывается по выходу блока кода, обрамляемого в with, так что в обработчике исключения файл был бы уже закрыт. Впрочем это обычно не существенное различие.
Еще пример:
# создадим виртуальную машину
with create_virtual_machine(root_passwd) as vm_ip:
# выполним на ней тестирования скрипта автоматической установки
test_auto_deploy_script(vm_ip, root_passwd)
# по выходу уничтожим vm_ip
# создадим виртуальную машину
with create_virtual_machine(root_passwd) as vm_ip:
# выполним на ней тестирования скрипта автоматической установки
test_auto_deploy_script(vm_ip, root_passwd)
# по выходу уничтожим vm_ip
- Захват/освобождение объекта Эту семантику поддерживают все стандартные объекты синхронизации
import threading
lock = Threading.Lock()
with lock:
# блокровка захваченна
pass
# блокировка отпущенна
import threading
lock = Threading.Lock()
with lock:
# блокровка захваченна
pass
# блокировка отпущенна
- Временное изменение настроек (примеры из документации python)
import warnings
from decimal import localcontext
with warnings.catch_warnings():
warnings.simplefilter("ignore")
# в этом участке кода все предепреждения игнорируются
with localcontext() as ctx:
ctx.prec = 42 # расчеты с типом Decimal выполняются с
# заоблачной точностью
s = calculate_something()
import warnings
from decimal import localcontext
with warnings.catch_warnings():
warnings.simplefilter("ignore")
# в этом участке кода все предепреждения игнорируются
with localcontext() as ctx:
ctx.prec = 42 # расчеты с типом Decimal выполняются с
# заоблачной точностью
s = calculate_something()
- Смена текущей директории (пример использования библиотеки fabric)
from fabric.context_managers import lcd
os.chdir('/opt')
print os.getcwd() # => /opt
with lcd('/tmp'):
print os.getcwd() # => /tmp
print os.getcwd() # => /opt
from fabric.context_managers import lcd
os.chdir('/opt')
print os.getcwd() # => /opt
with lcd('/tmp'):
print os.getcwd() # => /tmp
print os.getcwd() # => /opt
Нужно помнить, что изменение таким образом глобальных настроек в многопоточной программе может доставить много веселых минут при отладке.
- Подмена/восстановление объекта (временный monkey patching, пример использования библиотеки mock)
import mock
my_mock = mock.MagicMock()
with mock.patch('__builtin__.open', my_mock):
# open подменена на mock.MagicMock
with open('foo') as h:
pass
import mock
my_mock = mock.MagicMock()
with mock.patch('__builtin__.open', my_mock):
# open подменена на mock.MagicMock
with open('foo') as h:
pass
- Транзакции баз данных….
Менеджер транзакций для sqlalchemy
from config import DB_URI
from db_session import get_session
class DBWrapper(object):
def __init__(self):
self.session = None
def __enter__(self):
self.session = get_session(DB_URI)
def __exit__(self, exc, *args):
# при выходе из 'with':
if exc is None:
# если все прошло успешно коммитим
# транзакцию и закрываем курсор
self.session.commit()
# если было исключение - откатываем
self.session.close()
# тут методы, скрывающие работу с базой
with DBWrapper() as dbw: # открываем транзакцию
dbw.get_some_data()
dbw.update_some_data("...")
from config import DB_URI
from db_session import get_session
class DBWrapper(object):
def __init__(self):
self.session = None
def __enter__(self):
self.session = get_session(DB_URI)
def __exit__(self, exc, *args):
# при выходе из 'with':
if exc is None:
# если все прошло успешно коммитим
# транзакцию и закрываем курсор
self.session.commit()
# если было исключение - откатываем
self.session.close()
# тут методы, скрывающие работу с базой
with DBWrapper() as dbw: # открываем транзакцию
dbw.get_some_data()
dbw.update_some_data("...")
- ….и не только баз данных
from threading import local
import subprocess
# обобщенная транзакция - выполняет набор обратных действий
# при возникновении в блоке 'with' не обработанного исключения
class Transaction(object):
def __init__(self, parent):
self.rollback_cmds = []
self.set_parent(parent)
def set_parent(self, parent):
# родительская транзакция
# если откатывается родительская транзакция, то она автоматом
# откатывает и дочерние, даже если они было уже успешно закрыты
# если откатывается дочерняя, то родительская может продолжить
# исполнение, если код выше по стеку обработает исключение
if parent is not None:
self.parent_add = parent.add
else:
self.parent_add = lambda *cmd : None
def __enter__(self):
return self
def __exit__(self, exc, *dt):
if exc is None:
self.commit()
else:
self.rollback()
def add(self, cmd):
self.parent_add(cmd)
self.transaction.append(cmd)
def commit(self):
self.transaction = []
def rollback(self):
for cmd in reversed(self.transaction):
if isinstance(cmd, basestring):
subprocess.check_call(cmd, shell=True)
else:
cmd[0](*cmd[1:])
class AutoInheritedTransaction(object):
# словарь, id потока => [список вложенных транзакций]
# позволяет автоматически находить родительскую транзакцию
# в том случае, если для каждого потока может быть не более
# одной цепи вложенных транзакций
transactions = local()
def __init__(self):
super(AutoInheritedTransaction, self).__init__(self.current())
self.register()
def register(self):
self.transaction.list = getattr(self.transaction, 'list') + [self]
@classmethod
def current(cls):
return getattr(self.transaction, 'list', [None])[-1]
used_loop_devs = []
with AutoInheritedTransaction() as tr:
# создаем loop устройство
loop_name = subprocess.check_output("losetup -f --show /tmp/fs_image")
# вызов для его удаления
tr.add("losetup -d " + loop_name)
# записываем новое устройство в массив
used_loop_devs.append(loop_name)
tr.add(lambda : used_loop_devs.remove(
used_loop_devs.index(
loop_name)))
# монтируем его
subprocess.check_output("mount {0} /mnt/some_dir")
tr.add("umount /mnt/some_dir")
some_code
from threading import local
import subprocess
# обобщенная транзакция - выполняет набор обратных действий
# при возникновении в блоке 'with' не обработанного исключения
class Transaction(object):
def __init__(self, parent):
self.rollback_cmds = []
self.set_parent(parent)
def set_parent(self, parent):
# родительская транзакция
# если откатывается родительская транзакция, то она автоматом
# откатывает и дочерние, даже если они было уже успешно закрыты
# если откатывается дочерняя, то родительская может продолжить
# исполнение, если код выше по стеку обработает исключение
if parent is not None:
self.parent_add = parent.add
else:
self.parent_add = lambda *cmd : None
def __enter__(self):
return self
def __exit__(self, exc, *dt):
if exc is None:
self.commit()
else:
self.rollback()
def add(self, cmd):
self.parent_add(cmd)
self.transaction.append(cmd)
def commit(self):
self.transaction = []
def rollback(self):
for cmd in reversed(self.transaction):
if isinstance(cmd, basestring):
subprocess.check_call(cmd, shell=True)
else:
cmd[0](*cmd[1:])
class AutoInheritedTransaction(object):
# словарь, id потока => [список вложенных транзакций]
# позволяет автоматически находить родительскую транзакцию
# в том случае, если для каждого потока может быть не более
# одной цепи вложенных транзакций
transactions = local()
def __init__(self):
super(AutoInheritedTransaction, self).__init__(self.current())
self.register()
def register(self):
self.transaction.list = getattr(self.transaction, 'list') + [self]
@classmethod
def current(cls):
return getattr(self.transaction, 'list', [None])[-1]
used_loop_devs = []
with AutoInheritedTransaction() as tr:
# создаем loop устройство
loop_name = subprocess.check_output("losetup -f --show /tmp/fs_image")
# вызов для его удаления
tr.add("losetup -d " + loop_name)
# записываем новое устройство в массив
used_loop_devs.append(loop_name)
tr.add(lambda : used_loop_devs.remove(
used_loop_devs.index(
loop_name)))
# монтируем его
subprocess.check_output("mount {0} /mnt/some_dir")
tr.add("umount /mnt/some_dir")
some_code
Эта модель программирования позволяет группировать в одной точке код прямой и обратной операции и избавляет от вложенных try/finally. Также with предоставляет естественный интерфейс для STM. cpython-withatomic — один из вариантов STM для руthon с поддержкой with.
- Подавление исключений
def supress(*ex_types):
# стоит добавить логирования подавляемого исключения
try:
yield
except Exception as x:
if not isinstance(x, ex_types):
raise
with supress(OSError):
os.unlink("some_file")
def supress(*ex_types):
# стоит добавить логирования подавляемого исключения
try:
yield
except Exception as x:
if not isinstance(x, ex_types):
raise
with supress(OSError):
os.unlink("some_file")
- Генерация XML/HTML других структурированных языков.
from xmlbuilder import XMLBuilder
# новый xml документ
x = XMLBuilder('root')
x.some_tag
x.some_tag_with_data('text', a='12')
# вложенные теги
with x.some_tree(a='1'):
with x.data:
x.mmm
x.node(val='11')
print str(x) # <= string object
from xmlbuilder import XMLBuilder
# новый xml документ
x = XMLBuilder('root')
x.some_tag
x.some_tag_with_data('text', a='12')
# вложенные теги
with x.some_tree(a='1'):
with x.data:
x.mmm
x.node(val='11')
print str(x) # <= string object
Получим в итоге:
/>
a="12">text
a="1">
/>
val="11" />
text
Код библиотеки находится на xmlbuilder.
- Трассировка блока в логере (установка sys.settrace)
import sys
import contextlib
def on_event(fr, evt, data):
print fr, evt, data
return on_event
@contextlib.contextmanager
def trace_me():
prev_trace = sys.gettrace()
sys.settrace(on_event)
try:
yield
finally:
sys.settrace(prev_trace)
print "after finally"
with trace_me():
print "in with"
x = 1
y = 2
print "before gettrace"
sys.gettrace()
print "after gettrace"
import sys
import contextlib
def on_event(fr, evt, data):
print fr, evt, data
return on_event
@contextlib.contextmanager
def trace_me():
prev_trace = sys.gettrace()
sys.settrace(on_event)
try:
yield
finally:
sys.settrace(prev_trace)
print "after finally"
with trace_me():
print "in with"
x = 1
y = 2
print "before gettrace"
sys.gettrace()
print "after gettrace"
Этот
код напечатает:
in with
before gettrace
after gettrace
call None
line None
line None
line None
call None
line None
after finally
Для лучшего понимания трассировки питона — python-aware-python.
Ссылки:
www.python.org/dev/peps/pep-0343
docs.python.org/reference/compound_stmts.html#the-with-statement
github.com/koder-ua/megarepo/tree/master/xmlbuilder/xmlbuilder
www.voidspace.org.uk/python/mock/compare.html#mocking-a-context-manager
en.wikipedia.org/wiki/Monkey_patch
www.sqlalchemy.org
fabfile.org
en.wikipedia.org/wiki/Software_Transaction_Memory
bitbucket.org/arigo/cpython-withatomic
blip.tv/pycon-us-videos-2009-2010-2011/pycon-2011-python-aware-python-4896752
Исходники этого и других постов со скриптами лежат тут — github.com/koder-ua. При использовании их, пожалуйста, ссылайтесь на koder-ua.blogspot.com.
Автор: konstantin danilov
