xsetbg_db.py

   1 #! /usr/bin/env python3
   2 """XSetBg database
   3
   4 """
   5
   6 import os
   7 from sqlobject import SQLObject, connectionForURI, sqlhub, \
   8     UnicodeCol, IntCol, BoolCol, DatabaseIndex, dberrors
   9 from xsetbg_conf import xsetbg_conf
  10
  11 __all__ = ['xsetbg_db_path', 'xsetbg_db']
  12
  13 # octal; -rw-------; make the database file(s) readable only by the user
  14 os.umask(0o066)
  15
  16
  17 class XSetBg(SQLObject):
  18     full_name = UnicodeCol(alternateID=True)
  19     last_shown = IntCol(default=None)  # timestamp
  20     flag = BoolCol(default=None)
  21     is_image = BoolCol(default=None)
  22
  23     last_shown_idx = DatabaseIndex('last_shown')
  24     flag_idx = DatabaseIndex('flag')
  25     is_image_idx = DatabaseIndex('is_image')
  26
  27
  28 def recreate_db():
  29     global xsetbg_db
  30     if not xsetbg_db:
  31         xsetbg_db = XSetBg
  32         xsetbg_db.createTable()
  33     return xsetbg_db
  34
  35
  36 class SQLiteMassInsert(object):
  37     def __init__(self):
  38         self.connection = sqlhub.processConnection
  39
  40     def __enter__(self):
  41         connection = self.connection
  42         connection.query("PRAGMA synchronous=OFF")
  43         connection.query("PRAGMA count_changes=OFF")
  44         connection.query("PRAGMA journal_mode=MEMORY")
  45         connection.query("PRAGMA temp_store=MEMORY")
  46
  47         txn = connection.transaction()
  48         sqlhub.processConnection = txn
  49         return txn
  50
  51     def __exit__(self, *args, **kw):
  52         sqlhub.processConnection.commit(close=True)
  53         connection = sqlhub.processConnection = self.connection
  54         connection.query("VACUUM")
  55         connection.close()
  56
  57
  58 db_dirs = []
  59 try:
  60     xsetbg_db_path = xsetbg_conf.get('xsetbg', 'database')
  61 except Exception:
  62     xsetbg_db_path = None
  63
  64 if not xsetbg_db_path:
  65     if 'XDG_CACHE_HOME' in os.environ:
  66         db_dirs.append(os.environ['XDG_CACHE_HOME'])
  67     home_cache = os.path.expanduser('~/.cache')
  68     if home_cache not in db_dirs:
  69         db_dirs.append(home_cache)
  70     db_dirs.append(os.path.dirname(os.path.abspath(__file__)))
  71
  72     for d in db_dirs:
  73         xsetbg_db_path = os.path.join(d, 'xsetbg.sqlite')
  74         if os.path.exists(xsetbg_db_path):
  75             break
  76
  77 if xsetbg_db_path:
  78     sqlhub.processConnection = \
  79         connectionForURI('sqlite:///%s' % xsetbg_db_path)
  80     try:
  81         XSetBg.select()[0]
  82     except IndexError:
  83         xsetbg_db = XSetBg  # Table exists but is empty
  84     except dberrors.Error:
  85         xsetbg_db = None
  86     else:
  87         xsetbg_db = XSetBg
  88 else:
  89     xsetbg_db = None
  90
  91 if __name__ == '__main__':
  92     print("DB dirs:", db_dirs)
  93     print("DB file:", xsetbg_db_path)