Skip to content

Instantly share code, notes, and snippets.

@mheiber
Created September 19, 2015 23:09
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save mheiber/1ce6b854a97ab03b0510 to your computer and use it in GitHub Desktop.
Save mheiber/1ce6b854a97ab03b0510 to your computer and use it in GitHub Desktop.
connect to a database, execute multiple queries from a single string of SQL statements, get results as a list of ordered dictionaries
'''Python 2.7X and Python 3X'''
import sqlite3
import sys
import unittest
from collections import OrderedDict
class Database():
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
self.con.close()
def __init__(self, adaptor, DB):
with adaptor.connect(DB) as con:
self.con = con
self.cur = con.cursor()
def __raw_execute(self, sql_string, delimiter):
for statement in sql_string.split(delimiter):
if len(statement.strip()) > 0:
self.cur.execute(statement)
def __to_list_of_dictionaries(self):
cols = self.cur.description
column_names = [col[0] for col in cols]
lst = []
for values in self.cur:
d = OrderedDict(zip(column_names, values))
lst.append(d)
return lst
def execute(self, sql_string, delimiter=';'):
self.__raw_execute(sql_string, delimiter)
return self.__to_list_of_dictionaries()
class DatabaseTest(unittest.TestCase):
def test(self):
database_name = 'drive_durations.db'
statements = '''DROP TABLE IF EXISTS DURATIONS;
CREATE TABLE IF NOT EXISTS DURATIONS (
A INTEGER
);
INSERT INTO DURATIONS VALUES(1);
SELECT A FROM DURATIONS;'''
db = Database(sqlite3, database_name)
actual = db.execute(statements)
expected = [OrderedDict([('A', 1)])]
self.assertEqual(actual, expected)
if __name__ == '__main__':
unittest.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment