Skip to content

Instantly share code, notes, and snippets.

@jorgeluisrmx

jorgeluisrmx/Contets.md

Last active September 22, 2022 18:36
Show Gist options
  • Save jorgeluisrmx/db3abd7d9bb7c5064028fc6c7026e9cc to your computer and use it in GitHub Desktop.
Save jorgeluisrmx/db3abd7d9bb7c5064028fc6c7026e9cc to your computer and use it in GitHub Desktop.
Download ZIP
A set of snippets to manage SQLite databases using the singleton pattern
Raw
Contets.md

Contents

  • database.py - Singleton class for Sqlite DB management
  • query2dict.py - Method to return a query as a dictionary
Raw
database.py
import sqlite3
class Database(object):
DB_URI = None # db URI
CONN= None # db connection
CUR = None # db cursor
@classmethod
def initialize(cls, db_uri):
"""
Innitialize the connection with the sqlite DB
"""
if (cls.CONN is None) and (cls.CUR is None):
cls.DB_URI = db_uri
# chack same thread false to operate DB from different threads
cls.CONN = sqlite3.connect(db_uri, check_same_thread=False)
# enforce foreing key
cls.CONN.execute("PRAGMA foreign_keys = ON")
cls.CUR = cls.CONN.cursor()
else:
raise DatabaseError("Database has been already initialized. Close it before trying again")
@classmethod
def create_db_tables(cls):
"""
Creates a new CyclingLog SQLITE DB instance
"""
if cls.CONN:
# creating stages table
stages_table = """CREATE TABLE stages (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
name TEXT NOT NULL UNIQUE
);"""
cls.execute(stages_table)
# creating log table
activities_table = """CREATE TABLE activities (
date TEXT NOT NULL,
route TEXT NOT NULL,
mod TEXT,
stage INTEGER NOT NULL,
distance REAL NOT NULL,
time_start TEXT NOT NULL,
time INTEGER,
time_moving INTEGER,
speed REAL,
speed_moving REAL,
speed_max REAL,
elev_gain REAL,
elev_loss REAL,
elev_min REAL,
elev_max REAL,
cycle INTEGER NOT NULL,
toi INTEGER,
toi_label TEXT,
comments TEXT,
gpx_filename TEXT,
PRIMARY KEY(date, route),
FOREIGN KEY(stage) REFERENCES stages(id)
);"""
cls.execute(activities_table)
# commit changes
cls.commit()
else:
raise DatabaseError("Database has not been initialized")
@classmethod
def execute(cls, query, values=None):
"""
Execute a transaction in the database
"""
if cls.CONN:
if values:
return cls.CUR.execute(query, values)
return cls.CUR.execute(query)
else:
raise DatabaseError("Database has not been initialized")
@classmethod
def execute_many(cls, query, value_lst):
"""
Execute many transactions in the database
"""
if cls.CONN:
if not isinstance(value_lst, list):
raise DatabaseError("value_lst is not a list of tuples")
cls.CUR.executemany(query, value_lst)
cls.commit()
else:
raise DatabaseError("Database has not been initialized")
@classmethod
def commit(cls):
"""
Commit changes in database
"""
if cls.CONN:
cls.CONN.commit()
else:
raise DatabaseError("Database has not been initialized")
@classmethod
def close(cls):
"""
Closes database
"""
if cls.CONN:
cls.CONN.close()
cls.CONN, cls.CUR, cls.DB_URI = None, None, None
# ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~ ~
class DatabaseError(Exception):
pass
Raw
query2dict.py
from collections import OrderedDict
def sqlite_query2dict_list(qcur):
"""
Returns a list of ordered dictionaries containing the query results
:param qcur: query cursor
"""
header = [tup[0] for tup in qcur.description]
query_res = []
for row in qcur:
rowd = OrderedDict()
for key, value in zip(header, row):
rowd[key] = value
query_res.append(rowd)
return query_res
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment