Python for STEM Teachers
Oregon Curriculum Network

Context Managers in Python

Lets talk about "context managers" in Python.

The "with context( ) as ctx:" construct provides a way of doing setup and teardown behind the scenes.

The context.__enter__ method creates the context, whereas context.__exit__ cleans up the environment as we exit the scope.

A context is typically used to establish a connection with an ecosystem resource, such as a database. The file type object returned by open( ) is also natively a context manager, self-closing upon __exit__.

Lets start with a SQL_DB class that establishes a connection to a database and stores the resulting connection and cursor objects at the class level. The Python DB API gives us the wherewithall to talk to any SQL engine, provided an appropriate driver has been imported.

The Python standard library hase SQLite natively, giving us an ideal tool with which to prototype a generic SQL engine. It may actually be enough for your needs in some cases.

In [1]:
import time
import sqlite3 as sql
import os

import sys

class NoConnection(Exception):

class SQL_DB:  # a database
    # class level parameters
    backend  = 'sqlite3'
    user_initials  = 'KTU'
    timezone = int(time.strftime("%z", time.localtime()))
    target_path = "/Users/kurner/Documents/classroom_labs"  # current directory
    db_name = os.path.join(target_path, 'glossary.db')

    def mod_date():
        return time.mktime(time.gmtime())  # GMT time

    def connect(cls):
            if cls.backend == 'sqlite3':
                cls.conn = sql.connect(cls.db_name)  # connection
                cls.curs = cls.conn.cursor()         # cursor
            elif cls.backend == 'mysql': # not using this, gives idea
                cls.conn = sql.connect(host='localhost',
                                      user='root', port='8889')
                cls.curs = cls.conn.cursor()
        except Exception:
            raise NoConnection
    def disconnect(cls):

So far, this class hasn't done any work. We'll use the context manager to make the actual connection, using this DB object.

Upon initialization, this context manager expects a database class as an argument. Context managers may be created without arguments as well. The __enter__ method is then immediately executed, and what it returns (maybe None) becomes the object named by the optional "as extension", in this case the object named dbx.

Dot notation takes us into the self of the DBcontext instance (dbx) which in turn anchors db, the database class, which was passed in to __init__. Inside the database class are the connection and cursor objects required by the DB API.

DBcontext has no specific knowledge about internal tables, simply connects and disconnects. We can subclass this generic class if wishing to add more database-specific methods. We'll start with a Glossary subclass, then move to Elements.

In [2]:
class DBcontext:
    Generic parent class for connecting and disconnecting

    def __init__(self, db):
        self.db = db      # references database class
    def __enter__(self):
        return self       # <-- for use inside with scope

    def __exit__(self, *stuff_happens):
        if stuff_happens[0]:
            print("Exception raised!")
            return True # <-- if considered handled, otherwise False
        return True

class Glossary(DBcontext):
    Subclass with custom methods for this particular database
    def create_table(self):
        self.db.curs.execute("""DROP TABLE IF EXISTS Glossary""")
        self.db.curs.execute("""CREATE TABLE Glossary
            (gl_term text PRIMARY KEY,
             gl_definition text,
             updated_at int,
             updated_by text)""")

    def save_term(self, *the_data):
        query = ("INSERT INTO Glossary "
        "(gl_term, gl_definition, updated_at, updated_by) "
        "VALUES ('{}', '{}', {}, '{}')".format(*the_data))
        # print(query)

In the "with scope" below, we create a new table by calling a method saved inside the context manager itself. The a text file is read in and each line gets saved to a row in the Glossary table. Again, the context manager has a method for inserting rows.

In [3]:
with Glossary(SQL_DB) as dbx:  # <--- dbx returned by __enter__
    # for testing __exit__ in case of an exception
    # raise NoConnection
    FILE = os.path.join(dbx.db.target_path, "glossary.txt")
    with open(FILE, 'r', encoding='UTF-8') as gloss:
        lines = gloss.readlines()

    for line in lines:
        if len(line.strip()) == 0:
        term, definition = line.split(":", 1)
        right_now = dbx.db.mod_date()
        dbx.save_term(term[2:].strip(), definition.strip(), right_now, dbx.db.user_initials)

We now have a resuable context that will connect and disconnect from our database.

Lets use it to query the Glossary table...

In [4]:
with Glossary(SQL_DB) as dbx:
    query = "SELECT gl_term, gl_definition FROM Glossary ORDER BY gl_term"
    dbx.db.curs.execute(query)  # gets the data
    print("{:^80}".format("GLOSSARY OF TERMS"))
    print("-" * 80)
    print("Term                                |Abbreviated Definition   " )
    print("-" * 80)
    for term in dbx.db.curs.fetchmany(10):  # fetchone(), fetchmany(n), fetchall()
        print("{:35} | {:45}".format(term[0], term[1][:45]))

                               GLOSSARY OF TERMS                                
Term                                |Abbreviated Definition   
.NET                                | a virtual machine and the target runtime plat
AJAX                                | asynchronous JavaScript and XML.  Loosely des
API                                 | Application Program Interface.  Specifies how
ASCII                               | American Standard Code for Information Interc
Agile                               | a set of practices and work flows designed to
Apache                              | a free / open source web server, highly confi
Apache Foundation                   | an umbrella organization under which many ope
C                                   | the core language of operating systems develo
C#                                  | a Java-like language targeting the .NET virtu
C++                                 | C turned into an Object Oriented language.  M

OK, that was fun. We were able to obtain a partial listing of what's in the Glossary table. Connecting and disconnecting was handled behind the scenes.

Now lets subclass DBcontext once again and add methods specific to a Periodic Table of the Elements.

In [5]:
import chem_stuff

# modify database class to point to a different database file
SQL_DB.db_name = os.path.join(SQL_DB.target_path, 'periodic_table.db')

class ChemContext(DBcontext):
    Subclass with custom methods for this particular database
    def create_table(self):
        self.db.curs.execute("""DROP TABLE IF EXISTS Elements""")
        self.db.curs.execute("""CREATE TABLE Elements
            (elem_protons int PRIMARY KEY,
             elem_symbol text,
             elem_long_name text,
             elem_mass float,
             elem_series text,
             updated_at int,
             updated_by text)""")

    def save_term(self, *the_data):
        query = ("INSERT INTO Elements "
        "(elem_protons, elem_symbol, elem_long_name, elem_mass, elem_series,"
        "updated_at, updated_by) "
        "VALUES ({}, '{}', '{}', {}, '{}', {}, '{}')".format(*the_data))
        # print(query)
with ChemContext(SQL_DB) as dbx:
    FILE = os.path.join(dbx.db.target_path, "periodic_table.json")

    chem_stuff.load_elements(FILE)  # uses imported module to read JSON

    for atom in chem_stuff.all_elements.values():
        right_now = dbx.db.mod_date()
        dbx.save_term(atom.protons, atom.symbol, atom.long_name, atom.mass, atom.series,
                     right_now, dbx.db.user_initials)

File: /Users/kurner/Documents/classroom_labs/periodic_table.json loaded.

To test that our data table has indeed been populated, lets connect again and run a select query. We'll be able to print the rows directly. We're not bothering with an Atom class this time, just publishing row data directly from tuples.

In [6]:
with DBcontext(SQL_DB) as dbx:  # <--- dbx returned by __enter__
    query = ("SELECT elem_symbol, elem_long_name, elem_protons, elem_mass, elem_series" 
    " FROM Elements ORDER BY elem_protons")
    print("{:^70}".format("PERIODIC TABLE OF THE ELEMENTS"))
    print("-" * 70)
    print("Symbol |Long Name             |Protons |Mass   |Series  " )
    print("-" * 70)
    for the_atom in dbx.db.curs.fetchall():
        print("{:6} | {:20} | {:6} | {:5.2f} | {:15}".format(the_atom[0],

                    PERIODIC TABLE OF THE ELEMENTS                    
Symbol |Long Name             |Protons |Mass   |Series  
H      | Hydrogen             |      1 |  1.01 | Other nonmetal 
He     | Helium               |      2 |  4.00 | Nobel gas      
Li     | Lithium              |      3 |  6.94 | Alkali metal   
Be     | Beryllium            |      4 |  9.01 | Alkaline earth metal
B      | Boron                |      5 | 10.81 | Metalloid      
C      | Carbon               |      6 | 12.01 | Noble gas      
N      | Nitrogen             |      7 | 14.01 | Other nonmetal 
O      | Oxygen               |      8 | 16.00 | Other nonmetal 
F      | Fluorine             |      9 | 19.00 | Metalloid      
Ne     | Neon                 |     10 | 20.18 | Noble gas      
Na     | Sodium               |     11 | 22.99 | Alkali metal   
Mg     | Magnesium            |     12 | 24.30 | Alkaline earth metal
Al     | Aluminum             |     13 | 26.98 | Post-transition metal
Si     | Silicon              |     14 | 28.09 | Metalloid      
P      | Phosphorous          |     15 | 30.97 | Other nonmetal 
S      | Sulfur               |     16 | 32.06 | Other nonmetal 
Cl     | Chlorine             |     17 | 35.45 | Halogen        
Ar     | Argon                |     18 | 39.95 | Nobel gas      
K      | Potassium            |     19 | 39.10 | Alkali metal