Merge "SQlite storage improved."

marconi/storage/sqlite/controllers.py

@@ -36,11 +36,10 @@ class Queue(base.QueueBase):
         )''')
 
     def list(self, tenant):
-        ans = []
+        records = self.driver._run('''select name, metadata from Queues where
-        for rec in self.driver._run('''select name from Queues where
+                tenant = ?''', tenant)
-                tenant = ?''', tenant):
+        for k, v in records:
-            ans.append(rec[0])
+            yield {'name': k, 'metadata': v}
-        return ans
 
     def get(self, name, tenant):
         try:
@@ -48,7 +47,9 @@ class Queue(base.QueueBase):
                 self.driver._get('''select metadata from Queues where
                     tenant = ? and name = ?''', tenant, name)[0])
         except TypeError:
-            raise exceptions.DoesNotExist('/'.join([tenant, 'queues', name]))
+            msg = (_("Queue %(name)s does not exist for tenant %(tenant)s")
+                    % dict(name=name, tenant=tenant))
+            raise exceptions.DoesNotExist(msg)
 
     def upsert(self, name, metadata, tenant):
         with self.driver:
@@ -56,7 +57,7 @@ class Queue(base.QueueBase):
                     tenant = ? and name = ?''', tenant, name) is None
             self.driver._run('''replace into Queues values
                     (null, ?, ?, ?)''', tenant, name,
-                    json.dumps(metadata))
+                    json.dumps(metadata, ensure_ascii=False))
             return rc
 
     def delete(self, name, tenant):
@@ -97,19 +98,27 @@ class Message(base.MessageBase):
                 qid, = self.driver._get('''select id from Queues where
                         tenant = ? and name = ?''', tenant, queue)
             except TypeError:
-                raise exceptions.DoesNotExist(
+                msg = (_("Queue %(name)s does not exist for tenant %(tenant)s")
-                        '/'.join([tenant, 'queues', queue]))
+                        % dict(name=queue, tenant=tenant))
+                raise exceptions.DoesNotExist(msg)
+
+            # executemany() sets lastrowid to None, so no matter we manually
+            # generate the IDs or not, we still need to query for it.
             try:
-                newid = self.driver._get('''select id + 1 from Messages
+                unused, = self.driver._get('''select id + 1 from Messages
-                        where id = (select max(id) from Messages)''')[0]
+                        where id = (select max(id) from Messages)''')
             except TypeError:
-                newid = 1001
+                unused, = 1001,
-            for m in messages:
+
-                self.driver._run('''insert into Messages values
+            def it(newid):
-                        (?, ?, ?, ?, datetime())''',
+                for m in messages:
-                        newid, qid, m['ttl'], json.dumps(m))
+                    yield (newid, qid, m['ttl'],
-                newid += 1
+                            json.dumps(m, ensure_ascii=False))
-        return [str(x) for x in range(newid - len(messages), newid)]
+                    newid += 1
+
+            self.driver._run_multiple('''insert into Messages values
+                    (?, ?, ?, ?, datetime())''', it(unused))
+        return [str(x) for x in range(unused, unused + len(messages))]
 
     def delete(self, queue, message_id, tenant=None, claim=None):
         pass

marconi/storage/sqlite/driver.py

@@ -35,6 +35,9 @@ class Driver(storage.DriverBase):
     def _run(self, sql, *args):
         return self.__db.execute(sql, args)
 
+    def _run_multiple(self, sql, it):
+        self.__db.executemany(sql, it)
+
     def _get(self, sql, *args):
         return self._run(sql, *args).fetchone()
 

marconi/tests/storage/test_impl_sqlite.py

@@ -0,0 +1,23 @@
+# Copyright (c) 2013 Rackspace, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from marconi.storage import sqlite
+from marconi.storage.sqlite import controllers
+from marconi.tests.storage import base
+
+
+class SQliteQueueTests(base.QueueControllerTest):
+    driver_class = sqlite.Driver
+    controller_class = controllers.Queue

marconi/tests/test_sqlite.py

@@ -20,26 +20,13 @@ from marconi.storage import sqlite
 from marconi.tests import util as testing
 
 
+#TODO(zyuan): let tests/storage/base.py handle these
 class TestSqlite(testing.TestBase):
 
-    def test_sqlite(self):
+    def test_some_messages(self):
         storage = sqlite.Driver()
         q = storage.queue_controller
-        self.assertEquals(q.upsert('fizbit', {'_message_ttl': 40}, '480924'),
+        q.upsert('fizbit', {'_message_ttl': 40}, '480924')
-                True)
-        q.upsert('boomerang', {}, '480924')
-        q.upsert('boomerang', {}, '01314')
-        q.upsert('unrelated', {}, '01314')
-        self.assertEquals(set(q.list('480924')), set(['fizbit', 'boomerang']))
-        with testtools.ExpectedException(exceptions.DoesNotExist):
-            q.get('Fizbit', '480924')
-        self.assertEquals(q.upsert('fizbit', {'_message_ttl': 20}, '480924'),
-                False)
-        self.assertEquals(q.get('fizbit', '480924'), {'_message_ttl': 20})
-        q.delete('boomerang', '480924')
-        with testtools.ExpectedException(exceptions.DoesNotExist):
-            q.get('boomerang', '480924')
-
         m = storage.message_controller
         d = [
                 {"body": {