Mercurial > getan > getan
view getan/backend.py @ 23:9c4e8ba3c4fa
Added a new implementation of 'getan' based on urwid, a python console user interface library.
author | Ingo Weinzierl <ingo_weinzierl@web.de> |
---|---|
date | Sat, 28 Aug 2010 20:16:58 +0200 |
parents | |
children | f96a18c10836 |
line wrap: on
line source
# -*- coding: utf-8 -*- # # (c) 2008, 2009, 2010 by # Sascha L. Teichmann <sascha.teichmann@intevation.de> # Ingo Weinzierl <ingo.weinzierl@intevation.de> # # This is Free Software licensed unter the terms of GPLv3 or later. # For details see LICENSE coming with the source of 'getan'. # import logging import sqlite3 as db from getan.project import Project, Entry DEFAULT_DATABASE = "time.db" LOAD_ACTIVE_PROJECTS = ''' SELECT id, key, description, total FROM projects LEFT JOIN (SELECT project_id, sum(strftime('%s', stop_time) - strftime('%s', start_time)) AS total FROM entries GROUP BY project_id) ON project_id = id WHERE active ''' LOAD_PROJECT_ENTRIES = ''' SELECT id, project_id, start_time as "[timestamp]", stop_time as "[timestamp]", description FROM entries WHERE project_id = %i ORDER BY id DESC ''' INSERT_PROJECT_ENTRY = ''' INSERT INTO entries (project_id, start_time, stop_time, description) VALUES(?,?,?,?) ''' DELETE_PROJECT_ENTRY = 'DELETE FROM entries WHERE id = %i' MOVE_ENTRY = 'UPDATE entries SET project_id = ? WHERE id = ?' logger = logging.getLogger() class Backend: def __init__(self, database = DEFAULT_DATABASE): self.database = database self.con = db.connect(database, detect_types=db.PARSE_DECLTYPES | db.PARSE_COLNAMES) self.con.text_factory = lambda x: unicode(x, "utf-8", "ignore") def load_projects(self): """ Loads active projects from database and returns them as array """ logger.debug("load active projects from database.") cur = None try : cur = self.con.cursor() cur.execute(LOAD_ACTIVE_PROJECTS) projects = [] while True: row = cur.fetchone() if not row: break proj = Project(*row) proj.entries = self.load_entries(proj.id) projects.append(proj) logger.info("found %i active projects." % len(projects)) return projects finally: close(cur) def load_entries(self, project_id): """ Loads all entries that belong to a specific project """ logger.debug("load entries that belong to project %s" % project_id) cur = None try: cur = self.con.cursor() cur.execute(LOAD_PROJECT_ENTRIES % project_id) entries = [] while True: try: row = cur.fetchone() if not row: break entries.append(Entry(*row)) except: logger.warn("found invalid entry.") logger.debug("Found %i entries that belong to project '%i'" % (len(entries), project_id)) return entries finally: close(cur) def insert_project_entry(self, project, stop_time, desc): if project is None: return cur = None try: cur = self.con.cursor() cur.execute(INSERT_PROJECT_ENTRY, ( project.id, project.start, stop_time, desc)) self.con.commit() logger.debug("Added new entry '%s' of project '%s' into db" % (desc, project.desc)) project.entries = self.load_entries(project.id) finally: close(cur) def delete_entries(self, entries): if entries is None: return cur = None try: cur = self.con.cursor() for entry in entries: cur.execute(DELETE_PROJECT_ENTRY % entry.id) logger.debug("Deleted entry: %s (%d)" % (entry.desc, entry.id)) self.con.commit() finally: close(cur) def move_entries(self, entries, new_project_id): if entries is None or new_project_id is None: return cur = None try: cur = self.con.cursor() for entry in entries: cur.execute(MOVE_ENTRY, (new_project_id, entry.id)) logger.debug("Moved entry '%s' (id=%d) to project with id %d." % (entry.desc, entry.id, new_project_id)) self.con.commit() finally: close(cur) def close(cur): """ This function closes a database cursor if it is existing """ if cur: try: cur.close() except: logger.warn("could not close database cursor.") # vim:set ts=4 sw=4 si et sta sts=4 fenc=utf8: