Public API.

setup.py

@@ -34,5 +34,11 @@ setup(name='zodburi',
       tests_require = requires,
       install_requires = tests_require,
       test_suite="zodburi",
+      entry_points="""\
+      [zodburi.resolvers]
+      zeo = zodburi.resolvers:client_storage_resolver
+      file = zodburi.resolvers:file_storage_resolver
+      zconfig = zodburi.resolvers:zconfig_resolver
+      memory = zodburi.resolvers:mapping_storage_resolver
+      """
       )
-

zodburi/__init__.py

+from pkg_resources import iter_entry_points
+
+
+def resolve_uri(uri):
+    """
+    Returns a tuple, (factory, dbkw) where factory is a no-arg callable which
+    returns a storage matching the spec defined in the uri.  dbkw is a dict of
+    keyword arguments that may be passed to ZODB.DB.DB.
+    """
+    scheme = uri[:uri.find(':')]
+    for ep in iter_entry_points('zodburi.resolvers'):
+        if ep.name == scheme:
+            resolver = ep.load()
+            factory, dbkw = resolver(uri)
+            return factory, _get_dbkw(dbkw)
+    else:
+        raise KeyError('No resolver found for uri: %s' % uri)
+
+
+def _get_dbkw(kw):
+    dbkw = {
+        'cache_size': 10000,
+        'pool_size': 7,
+        'database_name': 'unnamed',
+    }
+    if 'connection_cache_size' in kw:
+        dbkw['cache_size'] = int(kw.pop('connection_cache_size'))
+    if 'connection_pool_size' in kw:
+        dbkw['pool_size'] = int(kw.pop('connection_pool_size'))
+    if 'database_name' in kw:
+        dbkw['database_name'] = kw.pop('database_name')
+
+    if kw:
+        raise KeyError('Unrecognized database keyword(s): %s' % ', '.join(kw))
+
+    return dbkw

zodburi/resolvers.py

@@ -8,7 +8,6 @@ from ZODB.FileStorage.FileStorage import FileStorage
 from ZODB.DemoStorage import DemoStorage
 from ZODB.MappingStorage import MappingStorage
 from ZODB.blob import BlobStorage
-from ZODB.DB import DB
 import ZConfig
 
 from zodburi.datatypes import convert_bytesize
@@ -124,7 +123,8 @@ class ClientStorageURIResolver(Resolver):
     _bytesize_args = ('cache_size', )
 
     def __call__(self, uri):
-        # urlparse doesnt understand zeo URLs so force to something that doesn't break
+        # urlparse doesnt understand zeo URLs so force to something that
+        # doesn't break
         uri = uri.replace('zeo://', 'http://', 1)
         (scheme, netloc, path, query, frag) = urlparse.urlsplit(uri)
         if netloc:
@@ -151,20 +151,6 @@ class ClientStorageURIResolver(Resolver):
                 return ClientStorage(*args, **kw)
         return factory, unused
 
-def get_dbkw(kw):
-    dbkw = {}
-    dbkw['cache_size'] = 10000
-    dbkw['pool_size'] = 7
-    dbkw['database_name'] = 'unnamed'
-    if 'connection_cache_size' in kw:
-        dbkw['cache_size'] = int(kw.pop('connection_cache_size'))
-    if 'connection_pool_size' in kw:
-        dbkw['pool_size'] = int(kw.pop('connection_pool_size'))
-    if 'database_name' in kw:
-        dbkw['database_name'] = kw.pop('database_name')
-
-    return dbkw
-
 
 class ZConfigURIResolver(object):
 
@@ -195,9 +181,7 @@ class ZConfigURIResolver(object):
         return factory.open, {}
 
 
-RESOLVERS = {
-    'zeo':ClientStorageURIResolver(),
-    'file':FileStorageURIResolver(),
-    'zconfig':ZConfigURIResolver(),
-    'memory':MappingStorageURIResolver(),
-    }
\ No newline at end of file
+client_storage_resolver = ClientStorageURIResolver()
+file_storage_resolver = FileStorageURIResolver()
+zconfig_resolver = ZConfigURIResolver()
+mapping_storage_resolver = MappingStorageURIResolver()
\ No newline at end of file

zodburi/tests/__init__.py

+import mock
+import unittest
+
+
+class TestResolveURI(unittest.TestCase):
+
+    @mock.patch('zodburi.resolvers.MappingStorage')
+    def test_it(self, MappingStorage):
+        from zodburi import resolve_uri
+        factory, dbkw = resolve_uri('memory://')
+        factory()
+        MappingStorage.assert_called_once_with('')
+        self.assertEqual(dbkw, {
+            'cache_size': 10000,
+            'pool_size': 7,
+            'database_name': 'unnamed'})
+
+    @mock.patch('zodburi.resolvers.MappingStorage')
+    def test_it_with_dbkw(self, MappingStorage):
+        from zodburi import resolve_uri
+        factory, dbkw = resolve_uri(
+            'memory://test?connection_cache_size=1&connection_pool_size=2&'
+            'database_name=dbname')
+        factory()
+        MappingStorage.assert_called_once_with('test')
+        self.assertEqual(dbkw, {
+            'cache_size': 1,
+            'pool_size': 2,
+            'database_name': 'dbname'})
+
+    def test_it_cant_resolve(self):
+        from zodburi import resolve_uri
+        self.assertRaises(KeyError, resolve_uri, 'http://whatevs')
+
+    def test_it_extra_kw(self):
+        from zodburi import resolve_uri
+        self.assertRaises(KeyError, resolve_uri, 'memory://?foo=bar')

zodburi/tests/test_resolvers.py

@@ -333,3 +333,19 @@ class TestMappingStorageURIResolver(Base, unittest.TestCase):
         from ZODB.MappingStorage import MappingStorage
         self.assertTrue(isinstance(storage, MappingStorage))
         self.assertEqual(storage.__name__, 'storagename')
+
+
+class TestEntryPoints(unittest.TestCase):
+
+    def test_it(self):
+        from pkg_resources import load_entry_point
+        from zodburi import resolvers
+        expected = [
+            ('memory', resolvers.MappingStorageURIResolver),
+            ('zeo', resolvers.ClientStorageURIResolver),
+            ('file', resolvers.FileStorageURIResolver),
+            ('zconfig', resolvers.ZConfigURIResolver),
+        ]
+        for name, cls in expected:
+            target = load_entry_point('zodburi', 'zodburi.resolvers', name)
+            self.assertTrue(isinstance(target, cls))