diff --git a/src/core/exceptions.py b/src/core/exceptions.py index 9b86484f..cb5ea9c5 100644 --- a/src/core/exceptions.py +++ b/src/core/exceptions.py @@ -75,3 +75,6 @@ class BridgeInitError(Exception): class BridgeExceptionNoService(Exception): pass + +class DatabaseError(Exception): + pass diff --git a/src/core/sat_main.py b/src/core/sat_main.py index 9864996e..12e31e80 100644 --- a/src/core/sat_main.py +++ b/src/core/sat_main.py @@ -82,15 +82,19 @@ class AbortSendMessage(Exception): class SAT(service.Service): + @property + def __version__(self): + return self.getConst('client_version') + def get_next_id(self): return sat_next_id() - def get_const(self, name): + def getConst(self, name): """Return a constant""" try: _const = os.environ['SAT_CONST_%s' % name] if _const: - debug(_("Constant %(name)s overrided with [%(value)s]") % {'name': name, 'value': _const}) + warning(_("Constant %(name)s overrided with [%(value)s]") % {'name': name, 'value': _const}) return _const except KeyError: pass @@ -99,7 +103,7 @@ class SAT(service.Service): raise Exception return CONST[name] - def set_const(self, name, value): + def setConst(self, name, value): """Save a constant""" if name in CONST: error(_('Trying to redefine a constant')) @@ -126,7 +130,7 @@ class SAT(service.Service): except exceptions.BridgeInitError: print (u"Bridge can't be initialised, can't start SàT core") # reactor is not launched yet, so we can't use error log sys.exit(1) - self.bridge.register("getVersion", lambda: self.get_const('client_version')) + self.bridge.register("getVersion", lambda: self.getConst('client_version')) self.bridge.register("getProfileName", self.memory.getProfileName) self.bridge.register("getProfilesList", self.memory.getProfilesList) self.bridge.register("getEntityData", lambda _jid, keys, profile: self.memory.getEntityData(jid.JID(_jid), keys, profile)) @@ -269,8 +273,8 @@ class SAT(service.Service): current.fallBack = xmpp.SatFallbackHandler(self) current.fallBack.setHandlerParent(current) - current.versionHandler = xmpp.SatVersionHandler(self.get_const('client_name'), - self.get_const('client_version')) + current.versionHandler = xmpp.SatVersionHandler(self.getConst('client_name'), + self.getConst('client_version')) current.versionHandler.setHandlerParent(current) current.identityHandler = xmpp.SatIdentityHandler() diff --git a/src/memory/memory.py b/src/memory/memory.py index 8a6bf12a..627506d4 100644 --- a/src/memory/memory.py +++ b/src/memory/memory.py @@ -782,10 +782,10 @@ class Memory(object): self.server_features = {} # used to store discovery's informations self.server_identities = {} self.config = self.parseMainConf() - host.set_const('savefile_database', SAVEFILE_DATABASE) + host.setConst('savefile_database', SAVEFILE_DATABASE) database_file = os.path.expanduser(self.getConfig('', 'local_dir') + - self.host.get_const('savefile_database')) - self.storage = SqliteStorage(database_file) + self.host.getConst('savefile_database')) + self.storage = SqliteStorage(database_file, host.__version__) PersistentDict.storage = self.storage self.params = Params(host, self.storage) info(_("Loading default params template")) diff --git a/src/memory/sqlite.py b/src/memory/sqlite.py index ff2a9d39..f321538a 100644 --- a/src/memory/sqlite.py +++ b/src/memory/sqlite.py @@ -18,18 +18,63 @@ # along with this program. If not, see . from sat.core.i18n import _ +from sat.core import exceptions from logging import debug, info, warning, error from twisted.enterprise import adbapi from twisted.internet import defer +from collections import OrderedDict from time import time +import re import os.path import cPickle as pickle +import hashlib + +CURRENT_DB_VERSION = 1 + +# XXX: DATABASE schemas are used in the following way: +# - 'current' key is for the actual database schema, for a new base +# - x(int) is for update needed between x-1 and x. All number are needed between y and z to do an update +# e.g.: if CURRENT_DB_VERSION is 6, 'current' is the actuel DB, and to update from version 3, numbers 4, 5 and 6 are needed +# a 'current' data dict can contains the keys: +# - 'CREATE': it contains an Ordered dict with table to create as keys, and a len 2 tuple as value, where value[0] are the columns definitions and value[1] are the table constraints +# - 'INSERT': it contains an Ordered dict with table where values have to be inserted, and many tuples containing values to insert in the order of the rows (#TODO: manage named columns) +# an update data dict (the ones with a number) can contains the keys 'create', 'delete', 'cols create', 'cols delete', 'cols modify' or 'insert'. See Updater.generateUpdateData for more infos. This metho can be used to autogenerate update_data, to ease the work of the developers. + +DATABASE_SCHEMAS = { + "current": {'CREATE': OrderedDict(( + ('profiles', (("id INTEGER PRIMARY KEY ASC", "name TEXT"), + ("UNIQUE (name)",))), + ('message_types', (("type TEXT PRIMARY KEY",), + tuple())), + ('history', (("id INTEGER PRIMARY KEY ASC", "profile_id INTEGER", "source TEXT", "dest TEXT", "source_res TEXT", "dest_res TEXT", "timestamp DATETIME", "message TEXT", "type TEXT", "extra BLOB"), + ("FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE", "FOREIGN KEY(type) REFERENCES message_types(type)"))), + ('param_gen', (("category TEXT", "name TEXT", "value TEXT"), + ("PRIMARY KEY (category,name)",))), + ('param_ind', (("category TEXT", "name TEXT", "profile_id INTEGER", "value TEXT"), + ("PRIMARY KEY (category,name,profile_id)", "FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE"))), + ('private_gen', (("namespace TEXT", "key TEXT", "value TEXT"), + ("PRIMARY KEY (namespace, key)",))), + ('private_ind', (("namespace TEXT", "key TEXT", "profile_id INTEGER", "value TEXT"), + ("PRIMARY KEY (namespace, key, profile_id)", "FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE"))), + ('private_gen_bin', (("namespace TEXT", "key TEXT", "value BLOB"), + ("PRIMARY KEY (namespace, key)",))), + + ('private_ind_bin', (("namespace TEXT", "key TEXT", "profile_id INTEGER", "value BLOB"), + ("PRIMARY KEY (namespace, key, profile_id)", "FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE"))) + )), + 'INSERT': OrderedDict(( + ('message_types', (("'chat'",), ("'error'",), ("'groupchat'",), ("'headline'",), ("'normal'",))), + )), + }, + 1: {'cols create': {'history': ('extra BLOB',)} + }, + } class SqliteStorage(object): """This class manage storage with Sqlite database""" - def __init__(self, db_filename): + def __init__(self, db_filename, sat_version): """Connect to the given database @param db_filename: full path to the Sqlite database""" self.initialized = defer.Deferred() # triggered when memory is fully initialised and ready @@ -38,31 +83,31 @@ class SqliteStorage(object): info(_("Connecting database")) new_base = not os.path.exists(db_filename) # do we have to create the database ? self.dbpool = adbapi.ConnectionPool("sqlite3", db_filename, check_same_thread=False) - + # init_defer is the initialisation deferred, initialisation is ok when all its callbacks have been done init_defer = self.dbpool.runOperation("PRAGMA foreign_keys = ON").addErrback(lambda x: error(_("Can't activate foreign keys"))) - if new_base: + def getNewBaseSql(): info(_("The database is new, creating the tables")) - database_creation = [ - "CREATE TABLE profiles (id INTEGER PRIMARY KEY ASC, name TEXT, UNIQUE (name))", - "CREATE TABLE message_types (type TEXT PRIMARY KEY)", - "INSERT INTO message_types VALUES ('chat')", - "INSERT INTO message_types VALUES ('error')", - "INSERT INTO message_types VALUES ('groupchat')", - "INSERT INTO message_types VALUES ('headline')", - "INSERT INTO message_types VALUES ('normal')", - "CREATE TABLE history (id INTEGER PRIMARY KEY ASC, profile_id INTEGER, source TEXT, dest TEXT, source_res TEXT, dest_res TEXT, timestamp DATETIME, message TEXT, type TEXT, extra BLOB, FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE, FOREIGN KEY(type) REFERENCES message_types(type))", - "CREATE TABLE param_gen (category TEXT, name TEXT, value TEXT, PRIMARY KEY (category,name))", - "CREATE TABLE param_ind (category TEXT, name TEXT, profile_id INTEGER, value TEXT, PRIMARY KEY (category,name,profile_id), FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE)", - "CREATE TABLE private_gen (namespace TEXT, key TEXT, value TEXT, PRIMARY KEY (namespace, key))", - "CREATE TABLE private_ind (namespace TEXT, key TEXT, profile_id INTEGER, value TEXT, PRIMARY KEY (namespace, key, profile_id), FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE)", - "CREATE TABLE private_gen_bin (namespace TEXT, key TEXT, value BLOB, PRIMARY KEY (namespace, key))", - "CREATE TABLE private_ind_bin (namespace TEXT, key TEXT, profile_id INTEGER, value BLOB, PRIMARY KEY (namespace, key, profile_id), FOREIGN KEY(profile_id) REFERENCES profiles(id) ON DELETE CASCADE)", - ] - for op in database_creation: - init_defer.addCallback(lambda ignore, sql: self.dbpool.runOperation(sql), op) - init_defer.addErrback(lambda ignore, sql: error(_("Error while creating tables in database [QUERY: %s]") % sql, op)) + database_creation = ["PRAGMA user_version=%d" % CURRENT_DB_VERSION] + database_creation.extend(Updater.createData2Raw(DATABASE_SCHEMAS['current']['CREATE'])) + database_creation.extend(Updater.insertData2Raw(DATABASE_SCHEMAS['current']['INSERT'])) + return database_creation + + def getUpdateSql(): + updater = Updater(self.dbpool, sat_version) + return updater.checkUpdates() + + def commitStatements(statements): + + if statements is None: + return defer.succeed(None) + debug("===== COMMITING STATEMENTS =====\n%s\n============\n\n" % '\n'.join(statements)) + d = self.dbpool.runInteraction(self._updateDb, tuple(statements)) + return d + + init_defer.addCallback(lambda ignore: getNewBaseSql() if new_base else getUpdateSql()) + init_defer.addCallback(commitStatements) def fillProfileCache(ignore): d = self.dbpool.runQuery("SELECT name,id FROM profiles").addCallback(self._profilesCache) @@ -70,6 +115,11 @@ class SqliteStorage(object): init_defer.addCallback(fillProfileCache) + def _updateDb(self, interaction, statements): + for statement in statements: + interaction.execute(statement) + + #Profiles def _profilesCache(self, profiles_result): """Fill the profiles cache @@ -402,3 +452,271 @@ class SqliteStorage(object): """Return the first result of a database query Useful when we are looking for one specific value""" return None if not result else result[0][0] + + +class Updater(object): + stmnt_regex = re.compile(r"(?:[\w ]+(?:\([\w, ]+\))?)+") + clean_regex = re.compile(r"^ +|(?<= ) +|(?<=,) +| +$") + CREATE_SQL = "CREATE TABLE %s (%s)" + INSERT_SQL = "INSERT INTO %s VALUES (%s)" + DROP_SQL = "DROP TABLE %s" + ALTER_SQL = "ALTER TABLE %s ADD COLUMN %s" + RENAME_TABLE_SQL = "ALTER TABLE %s RENAME TO %s" + + CONSTRAINTS = ('PRIMARY', 'UNIQUE', 'CHECK', 'FOREIGN') + TMP_TABLE = "tmp_sat_update" + + def __init__(self, dbpool, sat_version): + self._sat_version = sat_version + self.dbpool = dbpool + + def getLocalVersion(self): + """ Get local database version + @return: version (int) + + """ + return self.dbpool.runQuery("PRAGMA user_version").addCallback(lambda ret: int(ret[0][0])) + + def _setLocalVersion(self, version): + """ Set local database version + @param version: version (int) + @return: deferred + + """ + return self.dbpool.runOperation("PRAGMA user_version=%d" % version) + + def getLocalSchema(self): + """ return raw local schema + @return: list of strings with CREATE sql statements for local database + + """ + d = self.dbpool.runQuery("select sql from sqlite_master where type = 'table'") + d.addCallback(lambda result: [row[0] for row in result]) + return d + + @defer.inlineCallbacks + def checkUpdates(self): + """ Check is database schema update is needed, according to DATABASE_SCHEMAS + @return: deferred which fire a list of SQL update statements, or None if no update is needed + + """ + local_version = yield self.getLocalVersion() + raw_local_sch = yield self.getLocalSchema() + local_sch = self.rawStatements2data(raw_local_sch) + current_sch = DATABASE_SCHEMAS['current']['CREATE'] + local_hash = self.statementHash(local_sch) + current_hash = self.statementHash(current_sch) + + if local_hash == current_hash: + if local_version != CURRENT_DB_VERSION: + warning(_("Your local schema is up-to-date, but database versions mismatch, fixing it...")) + yield self._setLocalVersion(CURRENT_DB_VERSION) + else: + # an update is needed + + if local_version == CURRENT_DB_VERSION: + # Database mismatch and we have the latest version + if self._sat_version.endswith('D'): + # we are in a development version + update_data = self.generateUpdateData(local_sch, current_sch, False) + warning(_("There is a schema mismatch, but as we are on a dev version, database will be updated")) + update_raw = self.update2raw(update_data, True) + defer.returnValue(update_raw) + else: + error(_(u"schema version is up-to-date, but local schema differ from expected current schema")) + update_data = self.generateUpdateData(local_sch, current_sch, True) + warning(_(u"Here are the commands that should fix the situation, use at your own risk (do a backup before modifying database), you can go to SàT's MUC room at sat@chat.jabberfr.org for help\n### SQL###\n%s\n### END SQL ###\n") % u'\n'.join(("%s;" % statement for statement in self.update2raw(update_data)))) + raise exceptions.DatabaseError("Database mismatch") + else: + # Database is not up-to-date, we'll do the update + info(_("Database schema has changed, local database will be updated")) + update_raw = [] + for version in xrange(local_version+1, CURRENT_DB_VERSION+1): + try: + update_data = DATABASE_SCHEMAS[version] + except KeyError: + raise exceptions.InternalError("Missing update definition (version %d)" % version) + update_raw.extend(self.update2raw(update_data)) + update_raw.append("PRAGMA user_version=%d" % CURRENT_DB_VERSION) + defer.returnValue(update_raw) + + @staticmethod + def createData2Raw(data): + """ Generate SQL statements from statements data + @param data: dictionary with table as key, and statements data in tuples as value + @return: list of strings with raw statements + + """ + ret = [] + for table in data: + defs, constraints = data[table] + assert isinstance(defs, tuple) + assert isinstance(constraints, tuple) + ret.append(Updater.CREATE_SQL % (table, ', '.join(defs + constraints))) + return ret + + @staticmethod + def insertData2Raw(data): + """ Generate SQL statements from statements data + @param data: dictionary with table as key, and statements data in tuples as value + @return: list of strings with raw statements + + """ + ret = [] + for table in data: + values_tuple = data[table] + assert isinstance(values_tuple, tuple) + for values in values_tuple: + assert isinstance(values, tuple) + ret.append(Updater.INSERT_SQL % (table, ', '.join(values))) + return ret + + def statementHash(self, data): + """ Generate hash of template data + useful to compare schemas + + @param data: dictionary of "CREATE" statement, with tables names as key, + and tuples of (col_defs, constraints) as values + @return: hash as string + """ + hash_ = hashlib.sha1() + tables = data.keys() + tables.sort() + + def stmnts2str(stmts): + return ','.join([self.clean_regex.sub('',stmt) for stmt in sorted(stmts)]) + + for table in tables: + col_defs, col_constr = data[table] + hash_.update("%s:%s:%s" % (table, stmnts2str(col_defs), stmnts2str(col_constr))) + return hash_.digest() + + def rawStatements2data(self, raw_statements): + """ separate "CREATE" statements into dictionary/tuples data + @param raw_statements: list of CREATE statements as strings + @return: dictionary with table names as key, and a (col_defs, constraints) tuple + + """ + schema_dict = {} + for create_statement in raw_statements: + if not create_statement.startswith("CREATE TABLE "): + warning("Unexpected statement, ignoring it") + continue + _create_statement = create_statement[13:] + table, raw_col_stats = _create_statement.split(' ',1) + if raw_col_stats[0] != '(' or raw_col_stats[-1] != ')': + warning("Unexpected statement structure, ignoring it") + continue + col_stats = [stmt.strip() for stmt in self.stmnt_regex.findall(raw_col_stats[1:-1])] + col_defs = [] + constraints = [] + for col_stat in col_stats: + name = col_stat.split(' ',1)[0] + if name in self.CONSTRAINTS: + constraints.append(col_stat) + else: + col_defs.append(col_stat) + schema_dict[table] = (tuple(col_defs), tuple(constraints)) + return schema_dict + + def generateUpdateData(self, old_data, new_data, modify=False): + """ Generate data for automatic update between two schema data + @param old_data: data of the former schema (which must be updated) + @param new_data: data of the current schema + @param modify: if True, always use "cols modify" table, else try to ALTER tables + @return: update data, a dictionary with: + - 'create': dictionary of tables to create + - 'delete': tuple of tables to delete + - 'cols create': dictionary of columns to create (table as key, tuple of columns to create as value) + - 'cols delete': dictionary of columns to delete (table as key, tuple of columns to delete as value) + - 'cols modify': dictionary of columns to modify (table as key, tuple of old columns to transfert as value). With this table, a new table will be created, and content from the old table will be transfered to it, only cols specified in the tuple will be transfered. + + """ + + create_tables_data = {} + create_cols_data = {} + modify_cols_data = {} + delete_cols_data = {} + old_tables = set(old_data.keys()) + new_tables = set(new_data.keys()) + + def getChanges(set_olds, set_news): + to_create = set_news.difference(set_olds) + to_delete = set_olds.difference(set_news) + to_check = set_news.intersection(set_olds) + return tuple(to_create), tuple(to_delete), tuple(to_check) + + tables_to_create, tables_to_delete, tables_to_check = getChanges(old_tables, new_tables) + + for table in tables_to_create: + create_tables_data[table] = new_data[table] + + for table in tables_to_check: + old_col_defs, old_constraints = old_data[table] + new_col_defs, new_constraints = new_data[table] + for obj in old_col_defs, old_constraints, new_col_defs, new_constraints: + if not isinstance(obj, tuple): + raise InternalError("Columns definitions must be tuples") + defs_create, defs_delete, ignore = getChanges(set(old_col_defs), set(new_col_defs)) + constraints_create, constraints_delete, ignore = getChanges(set(old_constraints), set(new_constraints)) + created_col_names = set([name.split(' ',1)[0] for name in defs_create]) + deleted_col_names = set([name.split(' ',1)[0] for name in defs_delete]) + if (created_col_names.intersection(deleted_col_names or constraints_create or constraints_delete) or + (modify and (defs_create or constraints_create or defs_delete or constraints_delete))): + # we have modified columns, we need to transfer table + # we determinate which columns are in both schema so we can transfer them + old_names = set([name.split(' ',1)[0] for name in old_col_defs]) + new_names = set([name.split(' ',1)[0] for name in new_col_defs]) + modify_cols_data[table] = tuple(old_names.intersection(new_names)); + else: + if defs_create: + create_cols_data[table] = (defs_create) + if defs_delete or constraints_delete: + delete_cols_data[table] = (defs_delete) + + return {'create': create_tables_data, + 'delete': tables_to_delete, + 'cols create': create_cols_data, + 'cols delete': delete_cols_data, + 'cols modify': modify_cols_data + } + + def update2raw(self, update, dev_version=False): + """ Transform update data to raw SQLite statements + @param upadte: update data as returned by generateUpdateData + @param dev_version: if True, update will be done in dev mode: no deletion will be done, instead a message will be shown. This prevent accidental lost of data while working on the code/database. + @return: list of string with SQL statements needed to update the base + + """ + ret = self.createData2Raw(update.get('create', {})) + drop = [] + for table in update.get('delete', tuple()): + drop.append(self.DROP_SQL % table) + if dev_version: + if drop: + info("Dev version, SQL NOT EXECUTED:\n--\n%s\n--\n" % "\n".join(drop)) + else: + ret.extend(drop) + + cols_create = update.get('cols create', {}) + for table in cols_create: + for col_def in cols_create[table]: + ret.append(self.ALTER_SQL % (table, col_def)) + + cols_delete = update.get('cols delete', {}) + for table in cols_delete: + info("Following columns in table [%s] are not needed anymore, but are kept for dev version: %s" % (table, ", ".join(cols_delete[table]))) + + cols_modify = update.get('cols modify', {}) + for table in cols_modify: + ret.append(self.RENAME_TABLE_SQL % (table, self.TMP_TABLE)) + main, extra = DATABASE_SCHEMAS['current']['CREATE'][table] + ret.append(self.CREATE_SQL % (table, ', '.join(main + extra))) + common_cols = ', '.join(cols_modify[table]) + ret.append("INSERT INTO %s (%s) SELECT %s FROM %s" % (table, common_cols, common_cols, self.TMP_TABLE)) + ret.append(self.DROP_SQL % self.TMP_TABLE) + + insert = update.get('insert', {}) + ret.extend(self.insertData2Raw(insert)) + + return ret