diff --git a/src/leap/soledad/server/_blobs.py b/src/leap/soledad/server/_blobs.py
index 4a963e1054d041151818a04395ab2b22798b3ef0..cb726788744a76dc6c775497d94376a0161646e9 100644
--- a/src/leap/soledad/server/_blobs.py
+++ b/src/leap/soledad/server/_blobs.py
@@ -103,7 +103,7 @@ class FilesystemBlobsBackend(object):
def get_flags(self, user, blob_id, namespace=''):
path = self._get_path(user, blob_id, namespace)
if not os.path.isfile(path):
- return defer.fail(BlobNotFound)
+ return defer.fail(BlobNotFound())
if not os.path.isfile(path + '.flags'):
return defer.succeed([])
with open(path + '.flags', 'r') as flags_file:
@@ -113,13 +113,14 @@ class FilesystemBlobsBackend(object):
def set_flags(self, user, blob_id, flags, namespace=''):
path = self._get_path(user, blob_id, namespace)
if not os.path.isfile(path):
- raise BlobNotFound
+ return defer.fail(BlobNotFound())
for flag in flags:
if flag not in ACCEPTED_FLAGS:
- raise InvalidFlag(flag)
+ return defer.fail(InvalidFlag(flag))
with open(path + '.flags', 'w') as flags_file:
raw_flags = json.dumps(flags)
flags_file.write(raw_flags)
+ return defer.succeed(None)
@defer.inlineCallbacks
def write_blob(self, user, blob_id, fd, namespace=''):
@@ -143,7 +144,7 @@ class FilesystemBlobsBackend(object):
def delete_blob(self, user, blob_id, namespace=''):
blob_path = self._get_path(user, blob_id, namespace)
if not os.path.isfile(blob_path):
- raise BlobNotFound
+ return defer.fail(BlobNotFound())
self.__touch(blob_path + '.deleted')
os.unlink(blob_path)
try:
@@ -182,7 +183,8 @@ class FilesystemBlobsBackend(object):
elif order_by == '-date':
blob_ids.sort(key=lambda x: os.path.getmtime(x), reverse=True)
elif order_by:
- raise Exception("Unsupported order_by parameter: %s" % order_by)
+ exc = Exception("Unsupported order_by parameter: %s" % order_by)
+ return defer.fail(exc)
if filter_flag:
blob_ids = list(self._filter_flag(blob_ids, filter_flag))
blob_ids = [os.path.basename(path).replace('.deleted', '')
@@ -205,7 +207,7 @@ class FilesystemBlobsBackend(object):
def get_tag(self, user, blob_id, namespace=''):
blob_path = self._get_path(user, blob_id, namespace)
if not os.path.isfile(blob_path):
- raise BlobNotFound
+ return defer.fail(BlobNotFound())
with open(blob_path) as doc_file:
doc_file.seek(-16, 2)
tag = base64.urlsafe_b64encode(doc_file.read())
@@ -276,6 +278,48 @@ class BlobFile(resource.Resource):
return NOT_DONE_YET
+def _catchBlobNotFound(failure, request, user, blob_id):
+ failure.trap(BlobNotFound)
+ logger.error("Error 404: Blob %s does not exist for user %s"
+ % (blob_id, user))
+ request.setResponseCode(404)
+ request.write("Blob doesn't exists: %s" % blob_id)
+ request.finish()
+
+
+def _catchBlobExists(failure, request, user, blob_id):
+ failure.trap(BlobExists)
+ logger.error("Error 409: Blob %s already exists for user %s"
+ % (blob_id, user))
+ request.setResponseCode(409)
+ request.write("Blob already exists: %s" % blob_id)
+ request.finish()
+
+
+def _catchQuotaExceeded(failure, request, user):
+ failure.trap(QuotaExceeded)
+ logger.error("Error 507: Quota exceeded for user: %s" % user)
+ request.setResponseCode(507)
+ request.write('Quota Exceeded!')
+ request.finish()
+
+
+def _catchInvalidFlag(failure, request, user, blob_id):
+ failure.trap(InvalidFlag)
+ flag = failure.value.message
+ logger.error("Error 406: Attempted to set invalid flag %s for blob %s "
+ "for user %s" % (flag, blob_id, user))
+ request.setResponseCode(406)
+ request.write("Invalid flag: %s" % str(flag))
+ request.finish()
+
+
+def _catchAllErrors(self, e, request):
+ logger.error('Error processing request: %s' % e.getErrorMessage())
+ request.setResponseCode(500)
+ request.finish()
+
+
class BlobsResource(resource.Resource):
isLeaf = True
@@ -316,19 +360,14 @@ class BlobsResource(resource.Resource):
d.addCallback(lambda _: request.finish())
return NOT_DONE_YET
- def catchBlobNotFound(failure):
- failure.trap(BlobNotFound)
- request.setResponseCode(404)
- request.write("Blob doesn't exists: %s" % blob_id)
- request.finish()
-
only_flags = request.args.get('only_flags', [False])[0]
if only_flags:
d = self._handler.get_flags(user, blob_id, namespace)
- d.addErrback(catchBlobNotFound)
d.addCallback(lambda flags: json.dumps(flags))
d.addCallback(lambda flags: request.write(flags))
d.addCallback(lambda _: request.finish())
+ d.addErrback(_catchBlobNotFound, request, user, blob_id)
+ d.addErrback(_catchAllErrors, request)
return NOT_DONE_YET
d = self._handler.get_tag(user, blob_id, namespace)
@@ -339,47 +378,29 @@ class BlobsResource(resource.Resource):
namespace=namespace))
d.addCallback(lambda fd: BlobFile(fd))
d.addCallback(lambda res: res.render_GET(request))
+ d.addErrback(_catchBlobNotFound, request, user, blob_id)
+ d.addErrback(_catchAllErrors, request, finishRequest=True)
return NOT_DONE_YET
def render_DELETE(self, request):
logger.info("http put: %s" % request.path)
user, blob_id, namespace = self._validate(request)
-
- def catchBlobNotFound(failure):
- failure.trap(BlobNotFound)
- request.setResponseCode(404)
- request.write("Blob doesn't exists: %s" % blob_id)
- request.finish()
-
d = self._handler.delete_blob(user, blob_id, namespace=namespace)
d.addCallback(lambda _: request.finish())
- d.addErrback(catchBlobNotFound)
+ d.addErrback(_catchBlobNotFound, request, user, blob_id)
+ d.addErrback(_catchAllErrors, request)
return NOT_DONE_YET
def render_PUT(self, request):
logger.info("http put: %s" % request.path)
user, blob_id, namespace = self._validate(request)
-
- def catchBlobExists(failure):
- failure.trap(BlobExists)
- request.setResponseCode(409)
- request.write("Blob already exists: %s" % blob_id)
- request.finish()
-
- def catchQuotaExceeded(failure):
- failure.trap(QuotaExceeded)
- logger.error("Error 507: Quota exceeded for user: %s" % user)
- request.setResponseCode(507)
- request.write('Quota Exceeded!')
- request.finish()
-
fd = request.content
d = self._handler.write_blob(user, blob_id, fd, namespace=namespace)
d.addCallback(lambda _: request.finish())
- d.addErrback(catchBlobExists)
- d.addErrback(catchQuotaExceeded)
- d.addErrback(self._error, request)
+ d.addErrback(_catchBlobExists, request, user, blob_id)
+ d.addErrback(_catchQuotaExceeded, request, user)
+ d.addErrback(_catchAllErrors, request)
return NOT_DONE_YET
def render_POST(self, request):
@@ -387,32 +408,14 @@ class BlobsResource(resource.Resource):
user, blob_id, namespace = self._validate(request)
raw_flags = request.content.read()
flags = json.loads(raw_flags)
-
- def catchBlobNotFound(failure):
- failure.trap(BlobNotFound)
- request.setResponseCode(404)
- request.write("Blob doesn't exists: %s" % blob_id)
- request.finish()
-
- def catchInvalidFlag(failure):
- e = failure.trap(InvalidFlag)
- request.setResponseCode(406)
- flag = e.message
- request.write("Invalid flag: %s" % str(flag))
- request.finish()
-
d = self._handler.set_flags(user, blob_id, flags, namespace=namespace)
d.addCallback(lambda _: request.write(''))
d.addCallback(lambda _: request.finish())
- d.addErrback(catchBlobNotFound)
- d.addErrback(catchInvalidFlag)
+ d.addErrback(_catchBlobNotFound, request, user, blob_id)
+ d.addErrback(_catchInvalidFlag, request, user, blob_id)
+ d.addErrback(_catchAllErrors, request)
return NOT_DONE_YET
- def _error(self, e, request):
- logger.error('Error processing request: %s' % e.getErrorMessage())
- request.setResponseCode(500)
- request.finish()
-
def _validate(self, request):
for arg in request.postpath:
if arg and not VALID_STRINGS.match(arg):
diff --git a/src/leap/soledad/server/_streaming_resource.py b/src/leap/soledad/server/_streaming_resource.py
index 9b672107a2c72dafb51c5770635c524846c6c01f..59bb383e5f73fbf2cb5b911d0a0fb305f1aca8aa 100644
--- a/src/leap/soledad/server/_streaming_resource.py
+++ b/src/leap/soledad/server/_streaming_resource.py
@@ -24,7 +24,7 @@ import base64
from zope.interface import implementer
from twisted.internet.interfaces import IPushProducer
-from twisted.internet import task
+from twisted.internet import task, defer
from twisted.web.server import NOT_DONE_YET
from twisted.web.resource import Resource
@@ -61,12 +61,20 @@ class StreamingResource(Resource):
db = self._handler
raw_content = request.content.read()
blob_ids = json.loads(raw_content)
- paths = []
+ deferreds = []
for blob_id in blob_ids:
+
+ def _get_blob_info(blob_id, path):
+ d = db.get_blob_size(user, blob_id, namespace)
+ d.addCallback(lambda size: (blob_id, path, size))
+ return d
+
path = db._get_path(user, blob_id, namespace)
- size = db.get_blob_size(user, blob_id, namespace)
- paths.append((blob_id, path, size))
- DownstreamProducer(request, paths).start()
+ d = _get_blob_info(blob_id, path)
+ deferreds.append(d)
+ d = defer.gatherResults(deferreds)
+ d.addCallback(
+ lambda paths: DownstreamProducer(request, paths).start())
return NOT_DONE_YET