rwlaunchpad/plugins/rwstagingmgr/rift/tasklets/rwstagingmgr/server/handler.py

   1 #
   2 #   Copyright 2016 RIFT.IO Inc
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16 # Author(s): Varun Prasad
  17 # Creation Date: 09/28/2016
  18 #
  19
  20 import tornado.httpclient
  21 import tornado.web
  22 import tornadostreamform.multipart_streamer as multipart_streamer
  23 import logging
  24 import os
  25
  26 MB = 1024 * 1024
  27 GB = 1024 * MB
  28
  29 MAX_STREAMED_SIZE = 5 * GB
  30
  31 class HttpMessageError(Exception):
  32     def __init__(self, code, msg):
  33         self.code = code
  34         self.msg = msg
  35
  36
  37 class RequestHandler(tornado.web.RequestHandler):
  38     def options(self, *args, **kargs):
  39         pass
  40
  41     def set_default_headers(self):
  42         self.set_header('Access-Control-Allow-Origin', '*')
  43         self.set_header('Access-Control-Allow-Headers',
  44                         'Content-Type, Cache-Control, Accept, X-Requested-With, Authorization')
  45         self.set_header('Access-Control-Allow-Methods', 'POST, GET, PUT, DELETE')
  46
  47
  48 class StoreStreamerPart(multipart_streamer.MultiPartStreamer):
  49     """
  50     Create a Part streamer with a custom temp directory. Using the default
  51     tmp directory and trying to move the file to $RIFT_ARTIFACTS occasionally
  52     causes link errors. So create a temp directory within the staging area.
  53     """
  54     def __init__(self, store, *args, **kwargs):
  55         super().__init__(*args, **kwargs)
  56         self.store = store
  57
  58     def create_part(self, headers):
  59         return multipart_streamer.TemporaryFileStreamedPart(self, headers, tmp_dir=self.store.tmp_dir)
  60
  61
  62 @tornado.web.stream_request_body
  63 class UploadStagingHandler(RequestHandler):
  64     def initialize(self, store):
  65         """Initialize the handler
  66
  67         Arguments:
  68             log  - the logger that this handler should use
  69             loop - the tasklets ioloop
  70
  71         """
  72         self.log = logging.getLogger()
  73         self.store = store
  74
  75         self.part_streamer = None
  76
  77     @tornado.gen.coroutine
  78     def prepare(self):
  79         """Prepare the handler for a request
  80
  81         The prepare function is the first part of a request transaction. It
  82         creates a temporary file that uploaded data can be written to.
  83
  84         """
  85         if self.request.method != "POST":
  86             return
  87
  88         self.request.connection.set_max_body_size(MAX_STREAMED_SIZE)
  89
  90         # Retrieve the content type and parameters from the request
  91         content_type = self.request.headers.get('content-type', None)
  92         if content_type is None:
  93             raise tornado.httpclient.HTTPError(400, "No content type set")
  94
  95         content_type, params = tornado.httputil._parse_header(content_type)
  96
  97         if "multipart/form-data" != content_type.lower():
  98             raise tornado.httpclient.HTTPError(415, "Invalid content type")
  99
 100         # You can get the total request size from the headers.
 101         try:
 102             total = int(self.request.headers.get("Content-Length", "0"))
 103         except KeyError:
 104             self.log.warning("Content length header not found")
 105             # For any well formed browser request, Content-Length should have a value.
 106             total = 0
 107
 108         # And here you create a streamer that will accept incoming data
 109         self.part_streamer = StoreStreamerPart(self.store, total)
 110
 111
 112     @tornado.gen.coroutine
 113     def data_received(self, chunk):
 114         """When a chunk of data is received, we forward it to the multipart streamer."""
 115         self.part_streamer.data_received(chunk)
 116
 117     def post(self, staging_id):
 118         """Handle a post request
 119
 120         The function is called after any data associated with the body of the
 121         request has been received.
 122
 123         """
 124         try:
 125             # You MUST call this to close the incoming stream.
 126             self.part_streamer.data_complete()
 127             desc_parts = self.part_streamer.get_parts_by_name("file")
 128             if len(desc_parts) != 1:
 129                 raise tornado.httpclient.HTTPError(400, "File option not found")
 130
 131             binary_data = desc_parts[0]
 132             staging_area = self.store.get_staging_area(staging_id)
 133             filename = binary_data.get_filename()
 134             staging_area.model.name = filename
 135             staging_area.model.size = binary_data.get_size()
 136
 137             dest_file = os.path.join(staging_area.model.path, filename)
 138             binary_data.move(dest_file)
 139
 140             self.set_status(200)
 141             self.write(tornado.escape.json_encode({
 142                 "path": "/api/download/{}/{}".format(staging_id, filename)
 143                     }))
 144
 145         finally:
 146             self.part_streamer.release_parts()
 147             self.finish()
 148