diff options
author | Adrien Di Mascio <Adrien.DiMascio@logilab.fr> | 2008-07-24 14:34:48 +0200 |
---|---|---|
committer | Adrien Di Mascio <Adrien.DiMascio@logilab.fr> | 2008-07-24 14:34:48 +0200 |
commit | 1a4dcdc0f6b17410c97511f85cdc9211360f2825 (patch) | |
tree | 65b1149858ea701843fb7cb092e83704af49efa5 | |
parent | 599c946f9039e418fcde1c6dd75eb138b811d30c (diff) | |
download | logilab-common-1a4dcdc0f6b17410c97511f85cdc9211360f2825.tar.gz |
[adbh] supported_backends is now defined in lgc (was defined in RQL before)
Each FunctionDescr can now declare on which DB it is supported
-rw-r--r-- | adbh.py | 13 | ||||
-rw-r--r-- | test/unittest_db.py | 46 |
2 files changed, 57 insertions, 2 deletions
@@ -31,6 +31,7 @@ class metafunc(type): class FunctionDescr(object): __metaclass__ = metafunc + supported_backends = () rtype = None # None <-> returned type should be the same as the first argument aggregat = False minargs = 1 @@ -86,6 +87,7 @@ class _GenericAdvFuncHelper: """ # DBMS resources descriptors and accessors + backend_name = None # overriden in subclasses ('postgres', 'sqlite', etc.) needs_from_clause = False union_parentheses_support = True users_support = True @@ -155,7 +157,7 @@ class _GenericAdvFuncHelper: def system_database(self): """return the system database for the given driver""" raise NotImplementedError('not supported by this DBMS') - + def backup_command(self, dbname, dbhost, dbuser, dbpassword, backupfile, keepownership=True): """return a command to backup the given database""" @@ -276,6 +278,7 @@ def pgdbcmd(cmd, dbhost, dbuser): class _PGAdvFuncHelper(_GenericAdvFuncHelper): """Postgres helper, taking advantage of postgres SEQUENCE support """ + backend_name = 'postgres' # modifiable but should not be shared FUNCTIONS = _GenericAdvFuncHelper.FUNCTIONS.copy() @@ -386,6 +389,7 @@ class _SqliteAdvFuncHelper(_GenericAdvFuncHelper): An exception is raised when the functionality is not emulatable """ + backend_name = 'sqlite' # modifiable but should not be shared FUNCTIONS = _GenericAdvFuncHelper.FUNCTIONS.copy() @@ -411,6 +415,7 @@ class _SqliteAdvFuncHelper(_GenericAdvFuncHelper): class _MyAdvFuncHelper(_GenericAdvFuncHelper): """MySQL helper, taking advantage of postgres SEQUENCE support """ + backend_name = 'mysql' needs_from_clause = True ilike_support = False # insensitive search by default @@ -514,3 +519,9 @@ def get_adv_func_helper(driver): def register_function(driver, funcdef): ADV_FUNC_HELPER_DIRECTORY[driver].register_function(funcdef) +# this function should be called `register_function` but the other +# definition was defined prior to this one +def auto_register_function(funcdef): + """register the function `funcdef` on supported backends""" + for driver in funcdef.supported_backends: + register_function(driver, funcdef) diff --git a/test/unittest_db.py b/test/unittest_db.py index de86afd..9610ae8 100644 --- a/test/unittest_db.py +++ b/test/unittest_db.py @@ -7,7 +7,10 @@ import socket from logilab.common.testlib import TestCase, unittest_main from logilab.common.db import * from logilab.common.db import PREFERED_DRIVERS -from logilab.common.adbh import _SqliteAdvFuncHelper, _PGAdvFuncHelper +from logilab.common.adbh import (_GenericAdvFuncHelper, _SqliteAdvFuncHelper, + _PGAdvFuncHelper, _MyAdvFuncHelper, + FunctionDescr, get_adv_func_helper, + auto_register_function) class PreferedDriverTC(TestCase): @@ -132,6 +135,7 @@ class DBAPIAdaptersTC(TestCase): def setUp(self): """Memorize original PREFERED_DRIVERS""" self.old_drivers = PREFERED_DRIVERS['postgres'][:] + self.base_functions = dict(_GenericAdvFuncHelper.FUNCTIONS) self.host = 'crater.logilab.fr' self.db = 'gincotest2' self.user = 'adim' @@ -140,6 +144,10 @@ class DBAPIAdaptersTC(TestCase): def tearDown(self): """Reset PREFERED_DRIVERS as it was""" PREFERED_DRIVERS['postgres'] = self.old_drivers + _GenericAdvFuncHelper.FUNCTIONS = self.base_functions + _PGAdvFuncHelper.FUNCTIONS = dict(self.base_functions) + _MyAdvFuncHelper.FUNCTIONS = dict(self.base_functions) + _SqliteAdvFuncHelper.FUNCTIONS = dict(self.base_functions) def test_raise(self): self.assertRaises(UnknownDriver, get_dbapi_compliant_module, 'pougloup') @@ -185,5 +193,41 @@ class DBAPIAdaptersTC(TestCase): self.failUnless(isinstance(module.adv_func_helper, _SqliteAdvFuncHelper)) + def test_auto_register_funcdef(self): + class RANDOM(FunctionDescr): + supported_backends = ('postgres', 'sqlite',) + rtype = 'Float' + minargs = maxargs = 0 + name_mapping = {'postgres': 'RANDOM', + 'mysql': 'RAND', + 'sqlite': 'SQLITE_RANDOM'} + auto_register_function(RANDOM) + + pghelper = get_adv_func_helper('postgres') + mshelper = get_adv_func_helper('mysql') + slhelper = get_adv_func_helper('sqlite') + self.failUnless('RANDOM' in pghelper.FUNCTIONS) + self.failUnless('RANDOM' in slhelper.FUNCTIONS) + self.failIf('RANDOM' in mshelper.FUNCTIONS) + + + def test_funcname_with_different_backend_names(self): + class RANDOM(FunctionDescr): + supported_backends = ('postgres', 'mysql', 'sqlite') + rtype = 'Float' + minargs = maxargs = 0 + name_mapping = {'postgres': 'RANDOM', + 'mysql': 'RAND', + 'sqlite': 'SQLITE_RANDOM'} + auto_register_function(RANDOM) + + pghelper = get_adv_func_helper('postgres') + mshelper = get_adv_func_helper('mysql') + slhelper = get_adv_func_helper('sqlite') + self.assertEquals(pghelper.func_sqlname('RANDOM'), 'RANDOM') + self.assertEquals(mshelper.func_sqlname('RANDOM'), 'RAND') + self.assertEquals(slhelper.func_sqlname('RANDOM'), 'SQLITE_RANDOM') + + if __name__ == '__main__': unittest_main() |