# This file is part of jwmud, written by Helmut Pozimski in 2014. # # jwmud 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, version 2 of the License. # # jwmud 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 jwmud. If not, see . # -*- coding: utf8 -*- import sqlite3 from jwmudlib import jwmu_exceptions class db: def __init__(self, db_path): """Constructor, establishes the initial connection to the database.""" try: self.__connection = sqlite3.connect(db_path) except sqlite3.OperationalError: raise jwmu_exceptions.DataBaseAccessFailed() else: self.__cursor = self.__connection.cursor() def __del__(self): """Destructor, closes the connection to the database""" self.__connection.close() def __ExecuteQuery(self, query, arguments=None): """Executes any query and returns the result""" if arguments is None: query_result = self.__cursor.execute(query) else: query_result = self.__cursor.execute(query, arguments) return query_result def __ExecuteStatement(self, statement, arguments=None): """Executes any statements that perform a write access to the database """ try: if arguments is None: self.__cursor.execute(statement) else: self.__cursor.execute(statement, arguments) except sqlite3.OperationalError: raise jwmu_exceptions.DataBaseWriteFailed() else: self.__connection.commit() def Populate(self): """Creates the initial database structure with all the necessary tables. """ create_weekdays = """CREATE TABLE weekdays( day_id INTEGER PRIMARY KEY NOT NULL, alarm_time TEXT, name TEXT );""" create_categories = """CREATE TABLE categories( category_id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, name TEXT NOT NULL, alarm_time TEXT );""" create_alarmdays = """CREATE TABLE alarmdays( date TEXT NOT NULL, alarm_time TEXT, category INTEGER DEFAULT NULL, FOREIGN KEY(category) REFERENCES category(category_id), PRIMARY KEY(date, alarm_time) );""" # create the empty tables self.__ExecuteStatement(create_weekdays) self.__ExecuteStatement(create_categories) self.__ExecuteStatement(create_alarmdays) # fill the weekdays table with the existing 7 days of the week, # these are always hardcoded self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (1, 'undefined', 'Monday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (2, 'undefined', 'Tuesday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (3, 'undefined', 'Wednesday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (4, 'undefined', 'Thursday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (5, 'undefined', 'Friday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (6, 'undefined', 'Saturday');") self.__ExecuteStatement("INSERT INTO weekdays (day_id, alarm_time, " "name) VALUES (7, 'undefined', 'Sunday');") def FetchCategoryId(self, category_name): """Fetches the id of a category with a given name from the database and returns it. """ query = "SELECT category_id FROM categories WHERE name=?;" result = self.__ExecuteQuery(query, (category_name,)) category_id = result.fetchone() try: return int(category_id[0]) except TypeError: raise jwmu_exceptions.CategoryNotFound() def ChangeAlarmTimeWeekdays(self, alarm_time, start=None, end=None, day=None): """Changes the alarm time either for a single day or a range of days, like 1-5 for regular work days or 6-7 for week end. What mode is chosen depends on the parameters defined when the function is called. """ statement = "UPDATE weekdays SET alarm_time = ? WHERE day_id = ?;" if day is not None: self.__ExecuteStatement(statement, (alarm_time, day)) elif start is not None and end is not None: for day in range(start, end+1): self.__ExecuteStatement(statement, (alarm_time, day)) else: raise jwmu_exceptions.WrongParameters() def CreateCategory(self, category_name): """Inserts a new category into the database and returns the newly assigned category id. """ statement = "INSERT INTO categories (name) VALUES (?);" self.__ExecuteStatement(statement, (category_name,)) category_id = self.FetchCategoryId(category_name) return category_id def DeleteCategory(self, category_id): """Deletes a category with the given ID from the database.""" statement = "UPDATE alarmdays SET category=NULL WHERE category = ?;" self.__ExecuteStatement(statement, (category_id,)) statement = "DELETE FROM categories WHERE category_id = ?;" self.__ExecuteStatement(statement, (category_id,)) def InsertDay(self, date, alarm_time, category_id=None): """Inserts an additional day with the given parameters and optionally a category into the database. """ try: if category_id is not None: statement = "INSERT INTO alarmdays (date, alarm_time, category) " \ "VALUES (? , ?, ?);" self.__ExecuteStatement(statement, (date, alarm_time, category_id)) else: statement = "INSERT INTO alarmdays (date, alarm_time) VALUES " \ "(?, ?);" self.__ExecuteStatement(statement, (date, alarm_time)) except sqlite3.IntegrityError: raise jwmu_exceptions.DayAlreadyInDatabase() def ModifyAlarmTime(self, date, alarm_time): """Changes the alarm time for a given day in the database.""" statement = "UPDATE alarmdays SET alarm_time = ? WHERE date = ?;" self.__ExecuteStatement(statement, (alarm_time, date)) def DeleteDay(self, date): """Deletes a specified day from the database.""" statement = "DELETE FROM alarmdays WHERE date = ?;" self.__ExecuteStatement(statement, (date,)) def FetchDays(self): """Retrieves a list of all the days with their corresponding categories from the database. """ query = "SELECT date, alarmdays.alarm_time, name FROM alarmdays LEFT " \ "JOIN categories ON alarmdays.category=categories.category_id;" result = self.__ExecuteQuery(query).fetchall() return result def FetchWeekDays(self): """Retrieves a list of all the week days and their alarm times from the database. """ query = "SELECT * FROM weekdays;" result = self.__ExecuteQuery(query).fetchall() return result