openstack/zaqar
Code Issues Proposed changes

Merge "Fix sqlalchemy storage driver for Python 3"

Browse Source
This commit is contained in:
Jenkins 2014-06-03 09:56:41 +00:00 committed by Gerrit Code Review
commit 91d4a9084b
6 changed files with 25 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
marconi/queues/storage/sqlalchemy/claims.py
View File

@ -46,7 +46,7 @@ class ClaimController(storage.Claim):
'id': utils.msgid_encode(int(id)),
'ttl': ttl,
'age': (timeutils.utcnow() - created).seconds,
'body': body,
'body': utils.json_decode(body),
}
def get(self, queue, claim_id, project=None):

9
marconi/queues/storage/sqlalchemy/messages.py
View File

@ -14,7 +14,6 @@
# limitations under the License.
import calendar
import json
import sqlalchemy as sa
from sqlalchemy.sql import func as sfunc
@ -76,7 +75,7 @@ class MessageController(storage.Message):
'id': message_id,
'ttl': ttl,
'age': now - calendar.timegm(created.timetuple()),
'body': json.loads(body),
'body': utils.json_decode(body),
}
def bulk_get(self, queue, message_ids, project):
@ -110,7 +109,7 @@ class MessageController(storage.Message):
'id': utils.msgid_encode(int(id)),
'ttl': ttl,
'age': now - calendar.timegm(created.timetuple()),
'body': json.loads(body),
'body': utils.json_decode(body),
}
def first(self, queue, project=None, sort=1):
@ -203,7 +202,7 @@ class MessageController(storage.Message):
'id': utils.msgid_encode(id),
'ttl': ttl,
'age': now - calendar.timegm(created.timetuple()),
'body': json.loads(body),
'body': utils.json_decode(body),
}
yield it()
@ -229,7 +228,7 @@ class MessageController(storage.Message):
for m in messages:
yield dict(qid=qid,
ttl=m['ttl'],
body=json.dumps(m['body']),
body=utils.json_encode(m['body']),
client=str(client_uuid))
result = trans.execute(tables.Messages.insert(), list(it()))

10
marconi/queues/storage/sqlalchemy/queues.py
View File

@ -12,8 +12,6 @@
# License for the specific language governing permissions and limitations under
# the License.
import json
import sqlalchemy as sa
from sqlalchemy.sql import func as sfunc
@ -52,7 +50,7 @@ class QueueController(storage.Queue):
marker_name['next'] = rec[0]
yield ({'name': rec[0]} if not detailed
else
{'name': rec[0], 'metadata': json.loads(rec[1])})
{'name': rec[0], 'metadata': utils.json_decode(rec[1])})
yield it()
yield marker_name['next']
@ -66,7 +64,7 @@ class QueueController(storage.Queue):
tables.Queues.c.project == project,
tables.Queues.c.name == name
))
return json.loads(self.driver.get(sel)[0])
return utils.json_decode(self.driver.get(sel)[0])
except utils.NoResult:
raise errors.QueueDoesNotExist(name, project)
@ -76,7 +74,7 @@ class QueueController(storage.Queue):
try:
ins = tables.Queues.insert().values(project=project, name=name,
metadata=json.dumps({}))
metadata=utils.json_encode({}))
res = self.driver.run(ins)
except sa.exc.IntegrityError:
return False
@ -104,7 +102,7 @@ class QueueController(storage.Queue):
where(sa.and_(
tables.Queues.c.project == project,
tables.Queues.c.name == name)).\
values(metadata=json.dumps(metadata))
values(metadata=utils.json_encode(metadata))
res = self.driver.run(update)
try:

7
marconi/queues/storage/sqlalchemy/shards.py
View File

@ -23,7 +23,6 @@ Schema:
"""
import functools
import json
import sqlalchemy as sa
@ -71,7 +70,7 @@ class ShardsController(base.ShardsBase):
# TODO(cpp-cabrera): rename to upsert
@utils.raises_conn_error
def create(self, name, weight, uri, options=None):
opts = None if options is None else json.dumps(options)
opts = None if options is None else utils.json_encode(options)
try:
stmt = sa.sql.expression.insert(tables.Shards).values(
@ -104,7 +103,7 @@ class ShardsController(base.ShardsBase):
assert fields, '`weight`, `uri`, or `options` not found in kwargs'
if 'options' in fields:
fields['options'] = json.dumps(fields['options'])
fields['options'] = utils.json_encode(fields['options'])
stmt = sa.sql.update(tables.Shards).where(
tables.Shards.c.name == name).values(**fields)
@ -134,6 +133,6 @@ def _normalize(shard, detailed=False):
}
if detailed:
opts = shard[3]
ret['options'] = json.loads(opts) if opts else None
ret['options'] = utils.json_decode(opts) if opts else None
return ret

10
marconi/queues/storage/sqlalchemy/utils.py
View File

@ -19,7 +19,9 @@ import sqlalchemy as sa
from sqlalchemy import exc
from sqlalchemy.sql import func as sfunc
from marconi.openstack.common import jsonutils
from marconi.openstack.common import log as logging
from marconi.openstack.common import strutils
from marconi.queues.storage import errors
from marconi.queues.storage.sqlalchemy import tables
@ -121,3 +123,11 @@ def stat_message(message):
'age': message['age'],
'created': message['created'],
}
def json_encode(obj):
return strutils.safe_encode(jsonutils.dumps(obj), 'utf-8')
def json_decode(binary):
return jsonutils.loads(binary, 'utf-8')

5
tests/unit/queues/storage/test_impl_sqlalchemy.py
View File

@ -20,6 +20,7 @@ from marconi.queues.storage import errors
from marconi.queues.storage import sqlalchemy
from marconi.queues.storage.sqlalchemy import controllers
from marconi.queues.storage.sqlalchemy import tables
from marconi.queues.storage.sqlalchemy import utils
from marconi import tests as testing
from marconi.tests.queues.storage import base
@ -33,9 +34,9 @@ class SqlalchemyTableTests(testing.TestBase):
def test_table_queries(self):
self.engine.execute(tables.Queues.insert(), id=1, project='test',
name='marconi', metadata='aaaa')
name='marconi', metadata=utils.json_encode('aaaa'))
self.engine.execute(tables.Messages.insert(), id=1, qid=1, ttl=10,
body='bbbb', client='a',
body=utils.json_encode('bbbb'), client='a',
created=datetime.datetime.now())
self.engine.execute(tables.Claims.insert(), id=1, qid=1, ttl=10,
created=datetime.datetime.now())