Making a Python Context Manager
I’ve decided to get myself up-to-date with some Python features I rarely extend but often use.
One of them is the mighty context manager.
Context managers allow cleanup of resources that need some kind of closure after being used like files and threads(for those familiar with Java something similar is AutoCloseable interface).
More formally PEP 343 defines them as objects that implement a “context management protocol” which
is composed of __enter__()
and __exit__()
methods.
As a developer the best way to wrap my head about something is simply to go and implement it.
There are several options for implementing a context manager. As an OOP fan I went the class way. A very minimal context manager with no error handling looks as follows:
import sys
from io import IOBase
from typing import Tuple, List
class SimpleContextManager(object):
def __init__(self, file_name: str, method: str):
self.file_obj = open(file_name, method)
def __enter__(self) -> IOBase:
return self.file_obj
def __exit__(self, type, value, traceback):
self.file_obj.close()
return True
with SimpleContextManager(sys.argv[1], "r") as file:
for line in file:
print(line.strip())
When __enter__
is executed on with
line the result gets written into file
variable.
Any exception that gets thrown before __enter__
returns is up to the caller to solve.
If an exception gets thrown by file
line generator the exception is eaten inside the context manager.
This is why __exit__
has type, value, traceback arguments (think exc_type, exc_value, exc_traceback).
If return value of __exit__
is True
, an exception, thrown by file generator, would be eaten up.
If False
exception propagates out of with
body and caller needs to handler the error.
Now that we know a bit about making a simple context manager how about we create something a bit more useful.
I decided to make a basic CSV parser that can skip header without additional if statements.
Instead of returning file object from context manager I can return a custom iterator
that skips the header. Since __enter__
can return any object I can also return the header as a tuple.
I’ve first implemented an iterator over lines of file that skips header if configured:
class LineIterator(object):
def __init__(self,
file_object: IOBase,
has_header: bool = True,
separator: str = ";"):
self.file_object = file_object
self.has_header = has_header
self.separator = separator
self.first_line = True
def __next__(self) -> List[str]:
line = next(self.file_object)
if self.first_line:
next(self.file_object)
self.first_line = False
if line:
return line.strip().split(self.separator)
def __iter__(self) -> LineIterator:
return self
If the iterator receives has_header
as True it will skip the header line.
The final context manager then looks like the following:
class CsvFile(object):
def __init__(self,
file_name: str,
method: str,
has_header: bool = True,
separator: str = ";"):
self.has_header = has_header
self.separator = separator
self.file_name = file_name
self.method = method
self.file_obj = None
def __enter__(self) -> Tuple[List[str], LineIterator]:
if not self.file_obj:
try:
self.file_obj = open(self.file_name, self.method)
except Exception as e:
# if this is rethrown caller needs to handle the exception
logging.error("Failed opening file")
header = []
if self.has_header:
first_line = next(self.file_obj, None)
if first_line:
header = first_line.split(self.separator)
lines = LineIterator(self.file_obj, self.has_header, self.separator)
return header, lines
def __exit__(self, type, value, traceback):
if value:
logging.error("Failed processing CSV",
exc_info=(type, value, traceback))
if self.file_obj:
self.file_obj.close()
return True
If has_header
is passed in as True this context manager will on __enter__
return a tuple of headers and
a line iterator which will only iterate through non-header lines.
CsvFile
can be used in a with
statement as follows:
with CsvFile("test.csv", "r", has_header=True, separator=';') as (csv_header, csv_file):
for line in csv_file:
print(line)
Note that if the file doesn’t exist CsvFile
won’t throw. An exception will be generated inside the LineIterator
and logged at
logging.error("Failed processing CSV", exc_info=(type, value, traceback))
.
And that wraps up context managers. Full code samples are available on my GitHub.
Note: There are plenty of good CSV libraries out there so please avoid making your own.