2021年4月29日 星期四

[ Python 文章收集 ] The Curious Case of Python's Context Manager

 Source From Here

Preface
Discovering the quirks of Python's context manager

Python’s context managers are great for resource management and stopping the propagation of leaked abstractions. You’ve probably used it while opening a file or a database connection. Usually it starts with a with statement like this:
  1. with open("file.txt""wt") as f:  
  2.     f.write("contents go here")  
In the above case, file.txt gets automatically closed when the execution flow goes out of the scope. This is equivalent to writing:
  1. try:  
  2.     f = open("file.txt""wt")  
  3.     text = f.write("contents go here")  
  4. finally:  
  5.     f.close()  
Writing Custom Context Managers
To write a custom context manager, you need to create a class that includes the __enter__ and __exit__ methods. Let’s recreate a custom context manager that will execute the same workflow as above:
  1. class CustomFileOpen:  
  2.     """Custom context manager for opening files."""  
  3.   
  4.     def __init__(self, filename, mode):  
  5.         self.filename = filename  
  6.         self.mode = mode  
  7.   
  8.     def __enter__(self):  
  9.         self.f = open(self.filename, self.mode)  
  10.         return self.f  
  11.   
  12.     def __exit__(self, *args):  
  13.         self.f.close()  
You can use the above class just like a regular context manager:
  1. with CustomFileOpen("file.txt""wt") as f:  
  2.     f.write("contents go here")  
From Generators to Context Managers
Creating context managers by writing a class with __enter__ and __exit__ methods, is not difficult. However, you can achieve better brevity by defining them using contextlib.contextmanager decorator. This decorator converts a generator function into a context manager. The blueprint for creating context manager decorators goes something like this:
  1. @contextmanager  
  2. def some_generator(<arguments>):  
  3.     <setup>  
  4.     try:  
  5.         yield <value>  
  6.     finally:  
  7.         <cleanup>  
When you use the context manager with the with statement:
  1. with some_generator(<arguments>) as <variable>:  
  2.     <body>  
It roughly translates to:
  1. <setup>  
  2. try:  
  3.     <variable> = <value>  
  4.     <body>  
  5. finally:  
  6.     <cleanup>  
The setup code goes before the try..finally block. Notice the point where the generator yields. This is where the code block nested in the with statement gets executed. After the completion of the code block, the generator is then resumed. If an unhandled exception occurs in the block, it’s re-raised inside the generator at the point where the yield occurred and then the finally block is executed. If no unhandled exception occurs, the code gracefully proceeds to the finally block where you run your cleanup code.

Let’s implement the same CustomFileOpen context manager with contextmanager decorator:
  1. from contextlib import contextmanager  
  2.   
  3. @contextmanager  
  4. def CustomFileOpen(filename, method):  
  5.     """Custom context manager for opening a file."""  
  6.   
  7.     f = open(filename, method)  
  8.     try:  
  9.         yield f  
  10.   
  11.     finally:  
  12.         f.close()  
Writing Context Managers as Decorators
You can use context managers as decorators also. To do so, while defining the class, you have to inherit from contextlib.ContextDecorator class. Let’s make a RunTime decorator that will be applied on a file-opening function. The decorator will:
* Print a user provided description of the function
* Print the time it takes to run the function

  1. from contextlib import ContextDecorator  
  2. from time import time  
  3.   
  4. class RunTime(ContextDecorator):  
  5.     """Timing decorator."""  
  6.   
  7.     def __init__(self, description):  
  8.         self.description = description  
  9.   
  10.     def __enter__(self):  
  11.         print(self.description)  
  12.         self.start_time = time()  
  13.   
  14.     def __exit__(self, *args):  
  15.         self.end_time = time()  
  16.         run_time = self.end_time - self.start_time  
  17.         print(f"The function took {run_time} seconds to run.")  
You can use the decorator like this:
  1. @RunTime("This function opens a file")  
  2. def custom_file_write(filename, mode, content):  
  3.     with open(filename, mode) as f:  
  4.         f.write(content)  
Using the function like this should return:
>>> print(custom_file_write("file.txt", "wt", "jello"))
This function opens a file
The function took 0.0005390644073486328 seconds to run.
None

You can also create the same decorator via contextlib.contextmanager decorator.
  1. from contextlib import contextmanager  
  2.   
  3. @contextmanager  
  4. def runtime(description):  
  5.   
  6.     print(description)  
  7.     start_time = time()  
  8.     try:  
  9.         yield  
  10.     finally:  
  11.         end_time = time()  
  12.         run_time = end_time - start_time  
  13.         print(f"The function took {run_time} seconds to run.")  
Nesting Contexts
You can nest multiple context managers to manage resources simultaneously. Consider the following dummy manager:
  1. from contextlib import contextmanager  
  2.   
  3.   
  4. @contextmanager  
  5. def get_state(name):  
  6.     print("entering:", name)  
  7.     yield name  
  8.     print("exiting :", name)  
  9.   
  10.   
  11. # multiple get_state can be nested like this  
  12. with get_state("A") as A, get_state("B") as B, get_state("C") as C:  
  13.     print("inside with statement:", A, B, C)  
Output:
entering: A
entering: B
entering: C
inside with statement: A B C
exiting : C
exiting : B
exiting : A

Notice the order they’re closed. Context managers are treated as a stack, and should be exited in reverse order in which they’re entered. If an exception occurs, this order matters, as any context manager could suppress the exception, at which point the remaining managers will not even get notified of this. The __exit__ method is also permitted to raise a different exception, and other context managers then should be able to handle that new exception.

Combining Multiple Context Managers
You can combine multiple context managers too. Let’s consider these two managers:
  1. from contextlib import contextmanager  
  2.   
  3.   
  4. @contextmanager  
  5. def a(name):  
  6.     print("entering a:", name)  
  7.     yield name  
  8.     print("exiting a:", name)  
  9.   
  10.   
  11. @contextmanager  
  12. def b(name):  
  13.     print("entering b:", name)  
  14.     yield name  
  15.     print("exiting b:", name)  
Now combine these two using the decorator syntax. The following function takes the above define managers a and b and returns a combined context manager ab.
  1. @contextmanager  
  2. def ab(a, b):  
  3.     with a("A") as A, b("B") as B:  
  4.         yield (A, B)  
This can be used as:
  1. with ab(a, b) as AB:  
  2.     print("Inside the composite context manager:", AB)  
Output:
entering a: A
entering b: B
Inside the composite context manager: ('A', 'B')
exiting b: B
exiting a: A

If you have variable numbers of context managers and you want to combine them gracefully, contextlib.ExitStack is here to help. Let’s rewrite context manager ab using ExitStack. This function takes the individual context managers and their arguments as tuples and returns the combined manager:
  1. from contextlib import contextmanager, ExitStack  
  2.   
  3.   
  4. @contextmanager  
  5. def ab(cms, args):  
  6.     with ExitStack() as stack:  
  7.         yield [stack.enter_context(cm(arg)) for cm, arg in zip(cms, args)]  
Then you can use it this way:
  1. with ab((a, b), ("A""B")) as AB:  
  2.     print("Inside the composite context manager:", AB)  
Output:
entering a: A
entering b: B
Inside the composite context manager: ['A', 'B']
exiting b: B
exiting a: A

ExitStack can be also used in cases where you want to manage multiple resources gracefully. For example, suppose, you need to create a list from the contents of multiple files in a directory. Let’s see, how you can do so while avoiding accidental memory leakage with robust resource management:
  1. from contextlib import ExitStack  
  2. from pathlib import Path  
  3.   
  4. # ExitStack ensures all files are properly closed after o/p  
  5. with ExitStack() as stack:  
  6.     streams = (  
  7.         stack.enter_context(open(fname, "r")) for fname in Path("src").rglob("*.py")  
  8.     )  
  9.     contents = [f.read() for f in streams]  
Using Context Managers to Create SQLAlchemy Session
If you are familiar with SQLALchemy, Python’s SQL toolkit and Object Relational Mapper, then you probably know the usage of Session to run a query. A Session basically turns any query into a transaction and makes it atomic. Context managers can help you write a transaction session in a very elegant way. A basic querying workflow in SQLAlchemy may look like this:
  1. from sqlalchemy import create_engine  
  2. from sqlalchemy.orm import sessionmaker  
  3. from contextlib import contextmanager  
  4.   
  5. # an Engine, which the Session will use for connection resources  
  6. some_engine = create_engine("sqlite://")  
  7.   
  8. # create a configured "Session" class  
  9. Session = sessionmaker(bind=some_engine)  
  10.   
  11.   
  12. @contextmanager  
  13. def session_scope():  
  14.     """Provide a transactional scope around a series of operations."""  
  15.     session = Session()  
  16.     try:  
  17.         yield session  
  18.         session.commit()  
  19.     except:  
  20.         session.rollback()  
  21.         raise  
  22.     finally:  
  23.         session.close()  
The excerpt above creates an in memory SQLite connection and a session_scope function with context manager. The session_scope function takes care of committing and rolling back in case of exception automatically. The session_scope function can be used to run queries in the following way:
  1. with session_scope() as session:  
  2.     myobject = MyObject("foo""bar")  
  3.     session.add(myobject)  


[ Python 常見問題 ] When using unittest.mock.patch, why is autospec not True by default?

  Source From  Here Question When you patch a function using  mock , you have the option to specify  autospec  as True: If you set  autospec...