mylar/mylar/db.py

185 lines
6.5 KiB
Python
Executable File

# This file is part of Mylar.
#
# Mylar is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Mylar is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Mylar. If not, see <http://www.gnu.org/licenses/>.
#####################################
## Stolen from Sick-Beard's db.py ##
#####################################
from __future__ import with_statement
import os
import sqlite3
import threading
import time
import Queue
import mylar
from mylar import logger
db_lock = threading.Lock()
mylarQueue = Queue.Queue()
def dbFilename(filename="mylar.db"):
return os.path.join(mylar.DATA_DIR, filename)
class WriteOnly:
def __init__(self):
t = threading.Thread(target=self.worker, name="DB-WRITER")
t.daemon = True
t.start()
logger.fdebug('Thread WriteOnly initialized.')
def worker(self):
myDB = DBConnection()
#this should be in it's own thread somewhere, constantly polling the queue and sending them to the writer.
logger.fdebug('worker started.')
while True:
thisthread = threading.currentThread().name
if not mylarQueue.empty():
# Rename the main thread
logger.fdebug('[' + str(thisthread) + '] queue is not empty yet...')
(QtableName, QvalueDict, QkeyDict) = mylarQueue.get(block=True, timeout=None)
logger.fdebug('[REQUEUE] Table: ' + str(QtableName) + ' values: ' + str(QvalueDict) + ' keys: ' + str(QkeyDict))
sqlResult = myDB.upsert(QtableName, QvalueDict, QkeyDict)
if sqlResult:
mylarQueue.task_done()
return sqlResult
else:
time.sleep(1)
#logger.fdebug('[' + str(thisthread) + '] sleeping until active.')
class DBConnection:
def __init__(self, filename="mylar.db"):
self.filename = filename
self.connection = sqlite3.connect(dbFilename(filename), timeout=20)
self.connection.row_factory = sqlite3.Row
self.queue = mylarQueue
def fetch(self, query, args=None):
with db_lock:
if query == None:
return
sqlResult = None
attempt = 0
while attempt < 5:
try:
if args == None:
#logger.fdebug("[FETCH] : " + query)
cursor = self.connection.cursor()
sqlResult = cursor.execute(query)
else:
#logger.fdebug("[FETCH] : " + query + " with args " + str(args))
cursor = self.connection.cursor()
sqlResult = cursor.execute(query, args)
# get out of the connection attempt loop since we were successful
break
except sqlite3.OperationalError, e:
if "unable to open database file" in e.args[0] or "database is locked" in e.args[0]:
logger.warn('Database Error: %s' % e)
attempt += 1
time.sleep(1)
else:
logger.warn('DB error: %s' % e)
raise
except sqlite3.DatabaseError, e:
logger.error('Fatal error executing query: %s' % e)
raise
return sqlResult
def action(self, query, args=None):
with db_lock:
if query == None:
return
sqlResult = None
attempt = 0
while attempt < 5:
try:
if args == None:
#logger.fdebug("[ACTION] : " + query)
sqlResult = self.connection.execute(query)
else:
#logger.fdebug("[ACTION] : " + query + " with args " + str(args))
sqlResult = self.connection.execute(query, args)
self.connection.commit()
break
except sqlite3.OperationalError, e:
if "unable to open database file" in e.message or "database is locked" in e.message:
logger.warn('Database Error: %s' % e)
logger.warn('sqlresult: %s' % query)
attempt += 1
time.sleep(1)
else:
logger.error('Database error executing %s :: %s' % (query, e))
raise
return sqlResult
def select(self, query, args=None):
sqlResults = self.fetch(query, args).fetchall()
if sqlResults == None:
return []
return sqlResults
def selectone(self, query, args=None):
sqlResults = self.fetch(query, args)
if sqlResults == None:
return []
return sqlResults
def upsert(self, tableName, valueDict, keyDict):
thisthread = threading.currentThread().name
changesBefore = self.connection.total_changes
genParams = lambda myDict: [x + " = ?" for x in myDict.keys()]
query = "UPDATE " + tableName + " SET " + ", ".join(genParams(valueDict)) + " WHERE " + " AND ".join(genParams(keyDict))
self.action(query, valueDict.values() + keyDict.values())
if self.connection.total_changes == changesBefore:
query = "INSERT INTO " +tableName +" (" + ", ".join(valueDict.keys() + keyDict.keys()) + ")" + \
" VALUES (" + ", ".join(["?"] * len(valueDict.keys() + keyDict.keys())) + ")"
self.action(query, valueDict.values() + keyDict.values())
#else:
# logger.info('[' + str(thisthread) + '] db is currently locked for writing. Queuing this action until it is free')
# logger.info('Table: ' + str(tableName) + ' Values: ' + str(valueDict) + ' Keys: ' + str(keyDict))
# self.queue.put( (tableName, valueDict, keyDict) )
# #assuming this is coming in from a seperate thread, so loop it until it's free to write.
# #self.queuesend()