m_librarian/db.py

   1 #! /usr/bin/env python
   2
   3 __all__ = ['Author', 'Book', 'Extension', 'Genre', 'Language',
   4            'init_db', 'insert_name', 'update_counters',
   5            ]
   6
   7 import os
   8 from sqlobject import SQLObject, StringCol, UnicodeCol, IntCol, BoolCol, \
   9     ForeignKey, DateCol, DatabaseIndex, RelatedJoin, \
  10     connectionForURI, sqlhub, SQLObjectNotFound, dberrors
  11 from .config import ml_conf
  12
  13 try:
  14     db_uri = ml_conf.get('database', 'URI')
  15 except:
  16     db_uri = None
  17
  18 db_dirs = []
  19 if not db_uri:
  20     if 'XDG_CACHE_HOME' in os.environ:
  21         db_dirs.append(os.environ['XDG_CACHE_HOME'])
  22     home_cache = os.path.expanduser('~/.cache')
  23     if home_cache not in db_dirs:
  24         db_dirs.append(home_cache)
  25
  26     for d in db_dirs:
  27         db_file = os.path.join(d, 'm_librarian.sqlite')
  28         if os.path.exists(db_file):
  29             break
  30     else:
  31         # octal; -rw-------;
  32         # make the database file/directory readable/writeable only by the user
  33         os.umask(0066)
  34         db_dir = db_dirs[0]
  35         try:
  36             os.makedirs(db_dir)
  37         except OSError:  # Perhaps already exists
  38             pass
  39         db_file = os.path.join(db_dir, 'm_librarian.sqlite')
  40
  41     db_uri = 'sqlite://%s' % db_file.replace(os.sep, '/')
  42
  43
  44 sqlhub.processConnection = connection = connectionForURI(db_uri)
  45
  46 if connection.dbName == 'sqlite':
  47     # Speedup SQLite connection
  48     connection.query("PRAGMA synchronous=OFF")
  49     connection.query("PRAGMA count_changes=OFF")
  50     connection.query("PRAGMA journal_mode=MEMORY")
  51     connection.query("PRAGMA temp_store=MEMORY")
  52
  53
  54 class Author(SQLObject):
  55     name = UnicodeCol(unique=True)
  56     count = IntCol()
  57     count_idx = DatabaseIndex('count')
  58     books = RelatedJoin('Book', otherColumn='book_id')
  59
  60
  61 class Book(SQLObject):
  62     authors = RelatedJoin('Author')
  63     genres = RelatedJoin('Genre')
  64     title = UnicodeCol()
  65     series = UnicodeCol()
  66     ser_no = IntCol()
  67     archive = StringCol()
  68     file = StringCol()
  69     size = IntCol()
  70     lib_id = StringCol()
  71     deleted = BoolCol()
  72     extension = ForeignKey('Extension')
  73     date = DateCol()
  74     language = ForeignKey('Language')
  75     archive_file_idx = DatabaseIndex('archive', 'file', unique=True)
  76
  77
  78 class Extension(SQLObject):
  79     name = StringCol(unique=True)
  80     count = IntCol()
  81     count_idx = DatabaseIndex('count')
  82
  83
  84 class Genre(SQLObject):
  85     name = StringCol(unique=True)
  86     title = UnicodeCol()
  87     count = IntCol()
  88     count_idx = DatabaseIndex('count')
  89     books = RelatedJoin('Book', otherColumn='book_id')
  90
  91
  92 class Language(SQLObject):
  93     name = StringCol(unique=True)
  94     count = IntCol()
  95     count_idx = DatabaseIndex('count')
  96
  97
  98 def init_db():
  99     try:
 100         Book.select()[0]
 101     except IndexError:  # Table exists but is empty
 102         return
 103     except dberrors.Error:
 104         for table in Author, Extension, Genre, Language, Book:
 105             table.createTable()
 106     else:
 107         return
 108
 109
 110 def insert_name(table, name, **kw):
 111     try:
 112         return table.byName(name)
 113     except SQLObjectNotFound:
 114         return table(name=name, count=0, **kw)
 115
 116
 117 def update_counters():
 118     for author in Author.select():
 119         author.count = len(author.books)
 120
 121     for ext in Extension.select():
 122         ext.count = Book.select(Book.q.extension == ext.name).count()
 123
 124     for genre in Genre.select():
 125         genre.count = len(genre.books)
 126
 127     for language in Language.select():
 128         language.count = Book.select(Book.q.language == language.name).count()
 129
 130
 131 def test():
 132     print "DB dirs:", db_dirs
 133     if db_uri:
 134         print "DB URI:", db_uri
 135
 136 if __name__ == '__main__':
 137     test()