Json dataset via sqlite3

sqlite_dataset.py

import sqlite3


########################################
# simple json dataset via sqlite3
########################################
class StatusDataset(object):
    def __init__(self, db, name, create=True):
        self._db = db
        self._name = name
        if create:
            self._create_table()

    def _create_table(self):
        self._db._execute(
            "CREATE TABLE IF NOT EIXSTS %s (id INTEGER PRIMARY KEY, created_at REAL, data TEXT);" % self._name)

    def __iter__(self):
        cur = self._db._execute("SELECT id, created_at, data FROM %s;" % self._name)
        row = cur.fetchone()
        while row:
            yield row[0], row[1], json.loads(row[2])
            row = cur.fetchone()
        cur.close()

    @property
    def name(self):
        return self._name

    def get(self, since=None, until=None, limit=None, order='DESC'):
        sql = "SELECT id, created_at,data FROM %s" % self._name
        if since or until:
            sql += " WHERE"
            if since:
                sql += " created_at >= {}".format(since)
            if until:
                sql += " created_at <= {}".format(until)
        sql += " ORDER BY created_at {}".format(order)
        if limit:
            sql += " LIMIT {}".format(limit)
        sql += ";"
        with self._db._execute(sql) as cur:
            return [(rowid, created_at, json.loads(data)) for rowid, created_at, data in cur.fetchall()]

    def put(self, data, rowid=None):
        text = json.dumps(data)
        if rowid is None:
            now = time.time()
            cur = self._execute("INSERT INTO {} (created_at, data) VALUES ({}, '{}')".format(self._name, now, text))
            return cur.lastrowid
        else:
            self._execute("UPDATE {} SET data = '{}' WHERE id = {};".format(self._name, text, rowid))
            return rowid

    def destroy(self):
        self._db._execute("DROP TABLE IF EXISTS %s;" % self._name)


class StatusDB(object):
    def __init__(self, sqlite_uri):
        self.sqlite_uri = sqlite_uri
        self._connection = sqlite3.connect(self.sqlite_uri)

    def _execute(self, sql, auto_commit=True):
        cur = self._connection.execute(sql)
        if auto_commit:
            self._connection.commit()
        return cur

    def _query(self, sql):
        cur = self._connection.execute(sql)
        try:
            names = [t[0] for t in cur.descriptions]
            return [dict(zip(names, row)) for row in cur.fetchall()]
        finally:
            cur.close()

    def dataset_names(self):
        return [r['name'] for r in self._query("SELECT * FROM sqlite_master WHERE type='table';")]

    def dataset(self, dataset_name):
        return StatusDataset(self, dataset_name)