Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
# extract SQLite3 data to dict example, explained in this post:
# https://nickgeorge.net/programming/python-sqlite3-extract-to-dictionary/
# run with:
# python3 extract-to-dict.py
import os
import sqlite3
import sys
def sql_data_to_list_of_dicts(path_to_db, select_query):
"""Returns data from an SQL query as a list of dicts."""
try:
con = sqlite3.connect(path_to_db)
con.row_factory = sqlite3.Row
things = con.execute(select_query).fetchall()
unpacked = [{k: item[k] for k in item.keys()} for item in things]
return unpacked
except Exception as e:
print(f"Failed to execute. Query: {select_query}\n with error:\n{e}")
return []
finally:
con.close()
def make_database(path_to_db, schema, data):
"""create sample db and add data"""
try:
db = sqlite3.connect(path_to_db)
db.execute(schema)
db.commit()
for item in data:
db.execute(
"INSERT INTO data (fpath, n_measure, treatment, amplitude) VALUES(:fpath, :n_measure, :treatment, :amplitude)",
item,
)
db.commit() # commit after ever addition
print(f"Added data {item['fpath']}")
print("Done adding sample data")
except Exception as e:
print(f"Try deleting the database {path_to_db} and running again")
print(f"Exception is {e}")
finally:
db.commit()
db.close()
SAMPLE_DB = "temp_db.db"
SCHEMA = """CREATE TABLE data (fpath TEXT, n_measure INTEGER, treatment TEXT,
amplitude REAL)"""
TEST_DATA = [
{
"fpath": "path/to/file/one.dat",
"n_measure": 1,
"treatment": "Control",
"amplitude": 50.5,
},
{
"fpath": "path/to/file/two.dat",
"n_measure": 2,
"treatment": "Control",
"amplitude": 76.5,
},
{
"fpath": "path/to/file/three.dat",
"n_measure": 1,
"treatment": "Experimental",
"amplitude": 5.5,
},
]
if __name__ == "__main__":
make_database(SAMPLE_DB, SCHEMA, TEST_DATA)
# get it back out
returned_data = sql_data_to_list_of_dicts(SAMPLE_DB, "SELECT * FROM data")
if returned_data == TEST_DATA:
print("Success! returned_data == input data")
os.remove(SAMPLE_DB) # cleanup!
sys.exit(f"Removing database {SAMPLE_DB} and exiting.")
else:
print("Oops, something went wrong!!!")
print("Returned_data != input data")
print(f"Returned data: \n {returned_data}")
print(f"Input data: \n {TEST_DATA}")
os.remove(SAMPLE_DB)
sys.exit(f"Removing database {SAMPLE_DB} and exiting.")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment