with VAR = EXPR:
BLOCKVAR = EXPR
VAR.__enter__()
try:
BLOCK
finally:
VAR.__exit__()with f = open("/etc/passwd"):
BLOCK1
BLOCK2@contextmanager
def opening(filename):
f = open(filename)
try:
yield f
finally:
f.close()with f = opening(filename):
...read data from f...with VAR = EXPR:
BLOCK1with EXPR as VAR:
BLOCK1with EXPR as VAR:
BLOCKmgr = (EXPR)
exit = type(mgr).__exit__ # Not calling it yet
value = type(mgr).__enter__(mgr)
exc = True
try:
try:
VAR = value # Only if "as VAR" is present
BLOCK
except:
# The exceptional case is handled here
exc = False
if not exit(mgr, *sys.exc_info()):
raise
# The exception is swallowed if exit() returns true
finally:
# The normal and non-local-goto cases are handled here
if exc:
exit(mgr, None, None, None)from __future__ import with_statementclass GeneratorContextManager(object):
def __init__(self, gen):
self.gen = gen
def __enter__(self):
try:
return self.gen.next()
except StopIteration:
raise RuntimeError("generator didn't yield")
def __exit__(self, type, value, traceback):
if type is None:
try:
self.gen.next()
except StopIteration:
return
else:
raise RuntimeError("generator didn't stop")
else:
try:
self.gen.throw(type, value, traceback)
raise RuntimeError("generator didn't stop after throw()")
except StopIteration:
return True
except:
# only re-raise if it's *not* the exception that was
# passed to throw(), because __exit__() must not raise
# an exception unless __exit__() itself failed. But
# throw() has to raise the exception to signal
# propagation, so this fixes the impedance mismatch
# between the throw() protocol and the __exit__()
# protocol.
#
if sys.exc_info()[1] is not value:
raise
def contextmanager(func):
def helper(*args, **kwds):
return GeneratorContextManager(func(*args, **kwds))
return helper@contextmanager
def opening(filename):
f = open(filename) # IOError is untouched by GeneratorContext
try:
yield f
finally:
f.close() # Ditto for errors here (however unlikely)with locking(myLock):
BLOCKwith myLock:
BLOCKf = open(filename)
with f:
BLOCK1
with f:
BLOCK2- file
- thread.LockType
- threading.Lock
- threading.RLock
- threading.Condition
- threading.Semaphore
- threading.BoundedSemaphore@contextmanager
def locked(lock):
lock.acquire()
try:
yield
finally:
lock.release()with locked(myLock):
# Code here executes with myLock held. The lock is
# guaranteed to be released when the block is left (even
# if via return or by an uncaught exception).@contextmanager
def opened(filename, mode="r"):
f = open(filename, mode)
try:
yield f
finally:
f.close()with opened("/etc/passwd") as f:
for line in f:
print line.rstrip()@contextmanager
def transaction(db):
db.begin()
try:
yield None
except:
db.rollback()
raise
else:
db.commit()class locked:
def __init__(self, lock):
self.lock = lock
def __enter__(self):
self.lock.acquire()
def __exit__(self, type, value, tb):
self.lock.release()@contextmanager
def stdout_redirected(new_stdout):
save_stdout = sys.stdout
sys.stdout = new_stdout
try:
yield None
finally:
sys.stdout = save_stdoutwith opened(filename, "w") as f:
with stdout_redirected(f):
print "Hello world"@contextmanager
def opened_w_error(filename, mode="r"):
try:
f = open(filename, mode)
except IOError, err:
yield None, err
else:
try:
yield f, None
finally:
f.close()with opened_w_error("/etc/passwd", "a") as (f, err):
if err:
print "IOError:", err
else:
f.write("guido::0:0::/:/bin/sh\n")import signal
with signal.blocked():
# code executed without worrying about signalsimport decimal
@contextmanager
def extra_precision(places=2):
c = decimal.getcontext()
saved_prec = c.prec
c.prec += places
try:
yield None
finally:
c.prec = saved_precdef sin(x):
"Return the sine of x as measured in radians."
with extra_precision():
i, lasts, s, fact, num, sign = 1, 0, x, 1, x, 1
while s != lasts:
lasts = s
i += 2
fact *= i * (i-1)
num *= x * x
sign *= -1
s += num / fact * sign
# The "+s" rounds back to the original precision,
# so this must be outside the with-statement:
return +s@contextmanager
def localcontext(ctx=None):
"""Set a new local decimal context for the block"""
# Default to using the current context
if ctx is None:
ctx = getcontext()
# We set the thread context to a copy of this context
# to ensure that changes within the block are kept
# local to the block.
newctx = ctx.copy()
oldctx = decimal.getcontext()
decimal.setcontext(newctx)
try:
yield newctx
finally:
# Always restore the original context
decimal.setcontext(oldctx)from decimal import localcontext, ExtendedContext
def sin(x):
with localcontext() as ctx:
ctx.prec += 2
# Rest of sin calculation algorithm
# uses a precision 2 greater than normal
return +s # Convert result to normal precision
def sin(x):
with localcontext(ExtendedContext):
# Rest of sin calculation algorithm
# uses the Extended Context from the
# General Decimal Arithmetic Specification
return +s # Convert result to normal contextclass closing(object):
def __init__(self, obj):
self.obj = obj
def __enter__(self):
return self.obj
def __exit__(self, *exc_info):
try:
close_it = self.obj.close
except AttributeError:
pass
else:
close_it()# emulate opening():
with closing(open("argument.txt")) as contradiction:
for line in contradiction:
print line
# deterministically finalize an iterator:
with closing(iter(data_source)) as data:
for datum in data:
process(datum)class released:
def __init__(self, lock):
self.lock = lock
def __enter__(self):
self.lock.release()
def __exit__(self, type, value, tb):
self.lock.acquire()with my_lock:
# Operations with the lock held
with released(my_lock):
# Operations without the lock
# e.g. blocking I/O
# Lock is held again here@contextmanager
def nested(*contexts):
exits = []
vars = []
try:
try:
for context in contexts:
exit = context.__exit__
enter = context.__enter__
vars.append(enter())
exits.append(exit)
yield vars
except:
exc = sys.exc_info()
else:
exc = (None, None, None)
finally:
while exits:
exit = exits.pop()
try:
exit(*exc)
except:
exc = sys.exc_info()
else:
exc = (None, None, None)
if exc != (None, None, None):
# sys.exc_info() may have been
# changed by one of the exit methods
# so provide explicit exception info
raise exc[0], exc[1], exc[2]with nested(a, b, c) as (x, y, z):
# Perform operationwith a as x:
with b as y:
with c as z:
# Perform operation