diff --git a/bin/weedb/__init__.py b/bin/weedb/__init__.py index 0b1eaa33..5663991a 100644 --- a/bin/weedb/__init__.py +++ b/bin/weedb/__init__.py @@ -121,6 +121,16 @@ class Connection(object): should raise an exception of type weedb.ProgrammingError if the table does not exist.""" raise NotImplementedError + def get_variable(self, var_name): + """Return a database specific operational variable. Generally, things like + pragmas, or optimization-related variables. + + It returns a 2-way tuple: + (variable-name, variable-value) + If the variable does not exist, it returns None. + """ + raise NotImplemented + def begin(self): raise NotImplementedError diff --git a/bin/weedb/mysql.py b/bin/weedb/mysql.py index e92e98ed..cb096b57 100644 --- a/bin/weedb/mysql.py +++ b/bin/weedb/mysql.py @@ -169,6 +169,18 @@ class Connection(weedb.Connection): column_list = [row[1] for row in self.genSchemaOf(table)] return column_list + @guard + def get_variable(self, var_name): + cursor = self.connection.cursor() + try: + cursor.execute("SHOW VARIABLES LIKE '%s';" % var_name) + row = cursor.fetchone() + # This is actually a 2-way tuple (variable-name, variable-value), + # or None, if the variable does not exist. + return row + finally: + cursor.close() + @guard def begin(self): """Begin a transaction.""" @@ -182,7 +194,6 @@ class Connection(weedb.Connection): def rollback(self): self.connection.rollback() - class Cursor(object): """A wrapper around the MySQLdb cursor object""" diff --git a/bin/weedb/sqlite.py b/bin/weedb/sqlite.py index c04b8530..29143aad 100644 --- a/bin/weedb/sqlite.py +++ b/bin/weedb/sqlite.py @@ -153,6 +153,16 @@ class Connection(weedb.Connection): raise weedb.ProgrammingError("No such table %s" % table) return column_list + @guard + def get_variable(self, var_name): + cursor = self.connection.cursor() + try: + cursor.execute("PRAGMA %s;" % var_name) + row = cursor.fetchone() + return None if row is None else (var_name, row[0]) + finally: + cursor.close() + @guard def begin(self): self.connection.execute("BEGIN TRANSACTION") diff --git a/bin/weedb/test/test_weedb.py b/bin/weedb/test/test_weedb.py index 8f23830e..5e7211be 100644 --- a/bin/weedb/test/test_weedb.py +++ b/bin/weedb/test/test_weedb.py @@ -197,17 +197,34 @@ class TestSqlite(Common): self.db_dict = sqlite_db_dict super(TestSqlite, self).__init__(*args, **kwargs) + def test_variable(self): + weedb.create(self.db_dict) + _connect = weedb.connect(self.db_dict) + _v = _connect.get_variable('journal_mode') + self.assertEqual(_v[1].lower(), 'delete') + _v = _connect.get_variable('foo') + self.assertEqual(_v, None) + _connect.close() + class TestMySQL(Common): def __init__(self, *args, **kwargs): self.db_dict = mysql_db_dict super(TestMySQL, self).__init__(*args, **kwargs) - + + def test_variable(self): + weedb.create(self.db_dict) + _connect = weedb.connect(self.db_dict) + _v = _connect.get_variable('lower_case_table_names') + self.assertTrue(_v[1] in ['0', '1', '2'], "Unknown lower_case_table_names value") + _v = _connect.get_variable('foo') + self.assertEqual(_v, None) + _connect.close() def suite(): tests = ['test_drop', 'test_double_create', 'test_no_db', 'test_no_tables', 'test_create', 'test_bad_table', 'test_select', 'test_bad_select', - 'test_rollback', 'test_transaction'] + 'test_rollback', 'test_transaction', 'test_variable'] return unittest.TestSuite(map(TestSqlite, tests) + map(TestMySQL, tests)) if __name__ == '__main__':