graphenestorage/sqlite.py
# -*- coding: utf-8 -*-
import os
import sqlite3
import logging
from appdirs import user_data_dir
from .interfaces import StoreInterface
log = logging.getLogger(__name__)
timeformat = "%Y%m%d-%H%M%S"
class SQLiteFile:
"""This class ensures that the user's data is stored in its OS
preotected user directory:
**OSX:**
* `~/Library/Application Support/<AppName>`
**Windows:**
* `C:\\Documents and Settings\\<User>\\Application Data\\Local Settings\\<AppAuthor>\\<AppName>`
* `C:\\Documents and Settings\\<User>\\Application Data\\<AppAuthor>\\<AppName>`
**Linux:**
* `~/.local/share/<AppName>`
Furthermore, it offers an interface to generated backups
in the `backups/` directory every now and then.
.. note:: The file name can be overwritten when providing a keyword
argument ``profile``.
"""
def __init__(self, *args, **kwargs):
appauthor = "Fabian Schuh"
appname = kwargs.get("appname", "graphene")
data_dir = kwargs.get("data_dir", user_data_dir(appname, appauthor))
if "profile" in kwargs:
self.storageDatabase = "{}.sqlite".format(kwargs["profile"])
else:
self.storageDatabase = "{}.sqlite".format(appname)
self.sqlite_file = os.path.join(data_dir, self.storageDatabase)
""" Ensure that the directory in which the data is stored
exists
"""
if os.path.isdir(data_dir): # pragma: no cover
return
else: # pragma: no cover
os.makedirs(data_dir)
class SQLiteCommon(object):
"""This class abstracts away common sqlite3 operations.
This class should not be used directly.
When inheriting from this class, the following instance members must
be defined:
* ``sqlite_file``: Path to the SQLite Database file
"""
def sql_fetchone(self, query):
connection = sqlite3.connect(self.sqlite_file)
try:
cursor = connection.cursor()
cursor.execute(*query)
result = cursor.fetchone()
finally:
connection.close()
return result
def sql_fetchall(self, query):
connection = sqlite3.connect(self.sqlite_file)
try:
cursor = connection.cursor()
cursor.execute(*query)
results = cursor.fetchall()
finally:
connection.close()
return results
def sql_execute(self, query, lastid=False):
connection = sqlite3.connect(self.sqlite_file)
try:
cursor = connection.cursor()
cursor.execute(*query)
connection.commit()
except Exception:
connection.close()
raise
ret = None
try:
if lastid:
cursor = connection.cursor()
cursor.execute("SELECT last_insert_rowid();")
ret = cursor.fetchone()[0]
finally:
connection.close()
return ret
class SQLiteStore(SQLiteFile, SQLiteCommon, StoreInterface):
"""The SQLiteStore deals with the sqlite3 part of storing data into a
database file.
.. note:: This module is limited to two columns and merely stores
key/value pairs into the sqlite database
On first launch, the database file as well as the tables are created
automatically.
When inheriting from this class, the following three class members must
be defined:
* ``__tablename__``: Name of the table
* ``__key__``: Name of the key column
* ``__value__``: Name of the value column
"""
#:
__tablename__ = None
__key__ = None
__value__ = None
def __init__(self, *args, **kwargs):
#: Storage
SQLiteFile.__init__(self, *args, **kwargs)
StoreInterface.__init__(self, *args, **kwargs)
if self.__tablename__ is None or self.__key__ is None or self.__value__ is None:
raise ValueError("Values missing for tablename, key, or value!")
if not self.exists(): # pragma: no cover
self.create()
def _haveKey(self, key):
"""Is the key `key` available?"""
query = (
"SELECT {} FROM {} WHERE {}=?".format(
self.__value__, self.__tablename__, self.__key__
),
(key,),
)
return True if self.sql_fetchone(query) else False
def __setitem__(self, key, value):
"""Sets an item in the store
:param str key: Key
:param str value: Value
"""
if self._haveKey(key):
query = (
"UPDATE {} SET {}=? WHERE {}=?".format(
self.__tablename__, self.__value__, self.__key__
),
(value, key),
)
else:
query = (
"INSERT INTO {} ({}, {}) VALUES (?, ?)".format(
self.__tablename__, self.__key__, self.__value__
),
(key, value),
)
self.sql_execute(query)
def __getitem__(self, key):
"""Gets an item from the store as if it was a dictionary
:param str value: Value
"""
query = (
"SELECT {} FROM {} WHERE {}=?".format(
self.__value__, self.__tablename__, self.__key__
),
(key,),
)
result = self.sql_fetchone(query)
if result:
return result[0]
else:
if key in self.defaults:
return self.defaults[key]
else:
return None
def __iter__(self):
"""Iterates through the store"""
return iter(self.keys())
def keys(self):
query = ("SELECT {} from {}".format(self.__key__, self.__tablename__),)
return [x[0] for x in self.sql_fetchall(query)]
def __len__(self):
"""return lenght of store"""
query = ("SELECT id from {}".format(self.__tablename__),)
return len(self.sql_fetchall(query))
def __contains__(self, key):
"""Tests if a key is contained in the store.
May test againsts self.defaults
:param str value: Value
"""
if self._haveKey(key) or key in self.defaults:
return True
else:
return False
def items(self):
"""returns all items off the store as tuples"""
query = (
"SELECT {}, {} from {}".format(
self.__key__, self.__value__, self.__tablename__
),
)
r = []
for key, value in self.sql_fetchall(query):
r.append((key, value))
return r
def get(self, key, default=None):
"""Return the key if exists or a default value
:param str value: Value
:param str default: Default value if key not present
"""
if key in self:
return self.__getitem__(key)
else:
return default
# Specific for this library
def delete(self, key):
"""Delete a key from the store
:param str value: Value
"""
query = (
"DELETE FROM {} WHERE {}=?".format(self.__tablename__, self.__key__),
(key,),
)
self.sql_execute(query)
def wipe(self):
"""Wipe the store"""
query = ("DELETE FROM {}".format(self.__tablename__),)
self.sql_execute(query)
def exists(self):
"""Check if the database table exists"""
query = (
"SELECT name FROM sqlite_master " + "WHERE type='table' AND name=?",
(self.__tablename__,),
)
return True if self.sql_fetchone(query) else False
def create(self): # pragma: no cover
"""Create the new table in the SQLite database"""
query = (
(
"""
CREATE TABLE {} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
{} STRING(256),
{} STRING(256)
)"""
).format(self.__tablename__, self.__key__, self.__value__),
)
self.sql_execute(query)