openstack/zaqar
Code Issues Proposed changes

Implementation of the reference storage.

Browse Source 
Implements: blueprint storage-reference

Change-Id: I7b270208344e76a892e3bead764ba8b68fabca4d
This commit is contained in:
Zhihao Yuan 2013-03-14 14:50:45 -04:00 committed by Zhihao Yuan
parent a23643ec3c
commit 2b40336eeb
5 changed files with 211 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
marconi/kernel.py
View File

@ -14,7 +14,7 @@
# limitations under the License.
from marconi.common import config
from marconi.storage import reference as storage
from marconi.storage import sqlite as storage
from marconi.transport.wsgi import driver as wsgi

2
marconi/storage/reference/__init__.py → marconi/storage/sqlite/__init__.py
View File

@ -4,4 +4,4 @@ In-memory reference Storage Driver for Marconi.
Useful for automated testing and for prototyping storage driver concepts.
"""
from marconi.storage.reference.driver import Driver # NOQA
from marconi.storage.sqlite.driver import Driver # NOQA

115
marconi/storage/sqlite/controllers.py Normal file
View File

@ -0,0 +1,115 @@
# 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.
import json
from marconi.storage import base
from marconi.storage import exceptions
class Queue(base.QueueBase):
def __init__(self, driver):
self.driver = driver
self.driver._run('''create table if not exists Queues (
id INTEGER,
tenant TEXT,
name TEXT,
metadata TEXT,
PRIMARY KEY(id),
UNIQUE(tenant, name)
)''')
self.driver._run('''create unique index if not exists Paths on Queues (
tenant, name
)''')
def list(self, tenant):
ans = []
for rec in self.driver._run('''select name from Queues where
tenant = ?''', tenant):
ans.append(rec[0])
return ans
def get(self, name, tenant):
try:
return json.loads(
self.driver._get('''select metadata from Queues where
tenant = ? and name = ?''', tenant, name)[0])
except TypeError:
raise exceptions.DoesNotExist('/'.join([tenant, 'queues', name]))
def upsert(self, name, metadata, tenant):
with self.driver:
rc = self.driver._get('''select metadata from Queues where
tenant = ? and name = ?''', tenant, name) is None
self.driver._run('''replace into Queues values
(null, ?, ?, ?)''', tenant, name,
json.dumps(metadata))
return rc
def delete(self, name, tenant):
self.driver._run('''delete from Queues where
tenant = ? and name = ?''', tenant, name)
def stats(self, name, tenant):
return {'messages': self.driver._get('''select count(id)
from Messages where
qid = (select id from Queues where
tenant = ? and name = ?)''', tenant, name)[0],
'actions': 0}
def actions(self, name, tenant, marker=None, limit=10):
pass
class Message(base.MessageBase):
def __init__(self, driver):
self.driver = driver
self.driver._run('''create table if not exists Messages (
id INTEGER,
qid INTEGER,
ttl INTEGER,
content TEXT,
created DATETIME,
PRIMARY KEY(id),
FOREIGN KEY(qid) references Queues(id) on delete cascade
)''')
def get(self, queue, tenant=None, message_id=None,
marker=None, echo=False, client_uuid=None):
pass
def post(self, queue, messages, tenant):
with self.driver:
try:
qid, = self.driver._get('''select id from Queues where
tenant = ? and name = ?''', tenant, queue)
except TypeError:
raise exceptions.DoesNotExist(
'/'.join([tenant, 'queues', queue]))
try:
newid = self.driver._get('''select id + 1 from Messages
where id = (select max(id) from Messages)''')[0]
except TypeError:
newid = 1001
for m in messages:
self.driver._run('''insert into Messages values
(?, ?, ?, ?, datetime())''',
newid, qid, m['ttl'], json.dumps(m))
newid += 1
return [str(x) for x in range(newid - len(messages), newid)]
def delete(self, queue, message_id, tenant=None, claim=None):
pass

33
marconi/storage/reference/driver.py → marconi/storage/sqlite/driver.py
View File

@ -13,22 +13,45 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import sqlite3
from marconi.common import config
from marconi import storage
from marconi.storage.sqlite import controllers
cfg = config.namespace('drivers:storage:sqlite').from_options(
database=':memory:')
class Driver(storage.DriverBase):
def __init__(self):
self.__path = cfg.database
self.__conn = sqlite3.connect(self.__path)
self.__db = self.__conn.cursor()
self._run('''PRAGMA foreign_keys = ON''')
def _run(self, sql, *args):
return self.__db.execute(sql, args)
def _get(self, sql, *args):
return self._run(sql, *args).fetchone()
def __enter__(self):
self._run('begin immediate')
def __exit__(self, exc_type, exc_value, traceback):
self.__conn.commit()
@property
def queue_controller(self):
# TODO(kgriffs): Create base classes for controllers in common/
return None
return controllers.Queue(self)
@property
def message_controller(self):
# TODO(kgriffs): Create base classes for controllers in common/
return None
return controllers.Message(self)
@property
def claim_controller(self):
# TODO(kgriffs): Create base classes for controllers in common/
return None

66
marconi/tests/test_sqlite.py Normal file
View File

@ -0,0 +1,66 @@
# 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.
import testtools
from marconi.storage import exceptions
from marconi.storage import sqlite
from marconi.tests.util import suite
class TestSqlite(suite.TestSuite):
def test_sqlite(self):
storage = sqlite.Driver()
q = storage.queue_controller
self.assertEquals(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": {
"event": "BackupStarted",
"backupId": "c378813c-3f0b-11e2-ad92-7823d2b0f3ce"
},
'ttl': 30
},
{"body": {
"event": "BackupProgress",
"currentBytes": "0",
"totalBytes": "99614720"
},
'ttl': 10
}
]
n = q.stats('fizbit', '480924')['messages']
l1 = m.post('fizbit', d, '480924')
l2 = m.post('fizbit', d, '480924')
self.assertEquals([int(v) + 2 for v in l1], map(int, l2))
self.assertEquals(q.stats('fizbit', '480924')['messages'] - n, 4)
q.delete('fizbit', '480924')
with testtools.ExpectedException(exceptions.DoesNotExist):
m.post('fizbit', d, '480924')