Multi-threaded API v2 handler

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 #### Improvements
 - Updated `guessit` to version 3.0.0 ([#4244](https://github.com/pymedusa/Medusa/pull/4244))
+- Updated the API v2 endpoint to handle concurrent requests ([#4970](https://github.com/pymedusa/Medusa/pull/4970))
 
 #### Fixes
 - Fixed many release name parsing issues as a result of updating `guessit` ([#4244](https://github.com/pymedusa/Medusa/pull/4244))

medusa/server/api/v2/alias.py

@@ -21,7 +21,7 @@ class AliasHandler(BaseRequestHandler):
     #: allowed HTTP methods
     allowed_methods = ('GET', 'POST', 'PUT', 'DELETE')
 
-    def get(self, identifier, path_param):
+    def http_get(self, identifier, path_param):
         """Query scene_exception information."""
         cache_db_con = db.DBConnection('cache.db')
         sql_base = (b'SELECT '
@@ -92,7 +92,7 @@ def get(self, identifier, path_param):
 
         return self._ok(data=data)
 
-    def put(self, identifier, **kwargs):
+    def http_put(self, identifier, **kwargs):
         """Update alias information."""
         identifier = self._parse(identifier)
         if not identifier:
@@ -128,7 +128,7 @@ def put(self, identifier, **kwargs):
 
         return self._no_content()
 
-    def post(self, identifier, **kwargs):
+    def http_post(self, identifier, **kwargs):
         """Add an alias."""
         if identifier is not None:
             return self._bad_request('Alias id should not be specified')
@@ -159,7 +159,7 @@ def post(self, identifier, **kwargs):
         data['id'] = cursor.lastrowid
         return self._created(data=data, identifier=data['id'])
 
-    def delete(self, identifier, **kwargs):
+    def http_delete(self, identifier, **kwargs):
         """Delete an alias."""
         identifier = self._parse(identifier)
         if not identifier:

medusa/server/api/v2/alias_source.py

@@ -34,7 +34,7 @@ class AliasSourceHandler(BaseRequestHandler):
     #: allowed HTTP methods
     allowed_methods = ('GET', )
 
-    def get(self, identifier, path_param=None):
+    def http_get(self, identifier, path_param=None):
         """Query alias source information.
 
         :param identifier: source name
@@ -71,7 +71,7 @@ class AliasSourceOperationHandler(BaseRequestHandler):
     #: allowed HTTP methods
     allowed_methods = ('POST', )
 
-    def post(self, identifier):
+    def http_post(self, identifier):
         """Refresh all scene exception types."""
         types = {
             'local': 'custom_exceptions',

medusa/server/api/v2/auth.py

@@ -34,7 +34,7 @@ def prepare(self):
         """Prepare."""
         pass
 
-    def post(self, *args, **kwargs):
+    def http_post(self, *args, **kwargs):
         """Request JWT."""
         username = app.WEB_USERNAME
         password = app.WEB_PASSWORD
@@ -56,7 +56,7 @@ def post(self, *args, **kwargs):
         if username != submitted_username or password != submitted_password:
             return self._failed_login(error='Invalid credentials')
 
-        self._login(submitted_exp)
+        return self._login(submitted_exp)
 
     def _login(self, exp=86400):
         self.set_header('Content-Type', 'application/json')
@@ -65,7 +65,7 @@ def _login(self, exp=86400):
 
         log.info('{user} logged into the API v2', {'user': app.WEB_USERNAME})
         time_now = int(time.time())
-        self._ok(data={
+        return self._ok(data={
             'token': jwt.encode({
                 'iss': 'Medusa ' + text_type(app.APP_VERSION),
                 'iat': time_now,
@@ -78,8 +78,8 @@ def _login(self, exp=86400):
         })
 
     def _failed_login(self, error=None):
-        self._unauthorized(error=error)
         log.warning('{user} attempted a failed login to the API v2 from IP: {ip}', {
             'user': app.WEB_USERNAME,
             'ip': self.request.remote_ip
         })
+        return self._unauthorized(error=error)

medusa/server/api/v2/base.py

@@ -10,6 +10,7 @@
 import operator
 import traceback
 from builtins import object
+from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime
 
 from babelfish.language import Language
@@ -21,13 +22,17 @@
 
 from six import itervalues, string_types, text_type, viewitems
 
+from tornado.gen import coroutine
 from tornado.httpclient import HTTPError
 from tornado.httputil import url_concat
+from tornado.ioloop import IOLoop
 from tornado.web import RequestHandler
 
 log = BraceAdapter(logging.getLogger(__name__))
 log.logger.addHandler(logging.NullHandler())
 
+executor = ThreadPoolExecutor(thread_name_prefix='APIv2-Thread')
+
 
 class BaseRequestHandler(RequestHandler):
     """A base class used for shared RequestHandler methods."""
@@ -74,6 +79,75 @@ def prepare(self):
         else:
             return self._unauthorized('Invalid token.')
 
+    def async_call(self, name, *args, **kwargs):
+        """Call the actual HTTP method, if available."""
+        try:
+            method = getattr(self, 'http_' + name)
+        except AttributeError:
+            raise HTTPError(405, '{name} method is not allowed'.format(name=name.upper()))
+
+        def blocking_call():
+            try:
+                return method(*args, **kwargs)
+            except Exception as error:
+                self._handle_request_exception(error)
+
+        return IOLoop.current().run_in_executor(executor, blocking_call)
+
+    @coroutine
+    def head(self, *args, **kwargs):
+        """HEAD HTTP method."""
+        content = self.async_call('head', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
+    @coroutine
+    def get(self, *args, **kwargs):
+        """GET HTTP method."""
+        content = self.async_call('get', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
+    @coroutine
+    def post(self, *args, **kwargs):
+        """POST HTTP method."""
+        content = self.async_call('post', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
+    @coroutine
+    def delete(self, *args, **kwargs):
+        """DELETE HTTP method."""
+        content = self.async_call('delete', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
+    @coroutine
+    def patch(self, *args, **kwargs):
+        """PATCH HTTP method."""
+        content = self.async_call('patch', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
+    @coroutine
+    def put(self, *args, **kwargs):
+        """PUT HTTP method."""
+        content = self.async_call('put', *args, **kwargs)
+        if content is not None:
+            content = yield content
+        if not self._finished:
+            self.finish(content)
+
     def write_error(self, *args, **kwargs):
         """Only send traceback if app.DEVELOPER is true."""
         if app.DEVELOPER and 'exc_info' in kwargs:
@@ -83,10 +157,11 @@ def write_error(self, *args, **kwargs):
                 self.write(line)
             self.finish()
         else:
-            self._internal_server_error()
+            response = self._internal_server_error()
+            self.finish(response)
 
     def options(self, *args, **kwargs):
-        """Options."""
+        """OPTIONS HTTP method."""
         self._no_content()
 
     def set_default_headers(self):
@@ -105,7 +180,7 @@ def set_default_headers(self):
             allowed_methods += self.allowed_methods
         self.set_header('Access-Control-Allow-Methods', ', '.join(allowed_methods))
 
-    def api_finish(self, status=None, error=None, data=None, headers=None, stream=None, content_type=None, **kwargs):
+    def api_response(self, status=None, error=None, data=None, headers=None, stream=None, content_type=None, **kwargs):
         """End the api request writing error or data to http response."""
         content_type = content_type or 'application/json; charset=UTF-8'
         if headers is not None:
@@ -114,21 +189,23 @@ def api_finish(self, status=None, error=None, data=None, headers=None, stream=No
         if error is not None and status is not None:
             self.set_status(status)
             self.set_header('content-type', content_type)
-            self.finish({
+            return {
                 'error': error
-            })
+            }
         else:
             self.set_status(status or 200)
             if data is not None:
                 self.set_header('content-type', content_type)
-                self.finish(json.JSONEncoder(default=json_default_encoder).encode(data))
+                return json.JSONEncoder(default=json_default_encoder).encode(data)
             elif stream:
                 # This is mainly for assets
                 self.set_header('content-type', content_type)
-                self.finish(stream)
+                return stream
             elif kwargs and 'chunk' in kwargs:
                 self.set_header('content-type', content_type)
-                self.finish(kwargs)
+                return kwargs
+
+        return None
 
     @classmethod
     def _create_base_url(cls, prefix_url, resource_name, *args):
@@ -165,14 +242,15 @@ def create_app_handler(cls, base):
 
         return cls.create_url(base, cls.name, *(cls.identifier, cls.path_param)), cls
 
-    def _handle_request_exception(self, e):
-        if isinstance(e, HTTPError):
-            self.api_finish(e.code, e.message)
+    def _handle_request_exception(self, error):
+        if isinstance(error, HTTPError):
+            response = self.api_response(error.code, error.message)
+            self.finish(response)
         else:
-            super(BaseRequestHandler, self)._handle_request_exception(e)
+            super(BaseRequestHandler, self)._handle_request_exception(error)
 
     def _ok(self, data=None, headers=None, stream=None, content_type=None):
-        self.api_finish(200, data=data, headers=headers, stream=stream, content_type=content_type)
+        return self.api_response(200, data=data, headers=headers, stream=stream, content_type=content_type)
 
     def _created(self, data=None, identifier=None):
         if identifier is not None:
@@ -181,37 +259,37 @@ def _created(self, data=None, identifier=None):
                 location += '/'
 
             self.set_header('Location', '{0}{1}'.format(location, identifier))
-        self.api_finish(201, data=data)
+        return self.api_response(201, data=data)
 
     def _accepted(self):
-        self.api_finish(202)
+        return self.api_response(202)
 
     def _no_content(self):
-        self.api_finish(204)
+        return self.api_response(204)
 
     def _multi_status(self, data=None, headers=None):
-        self.api_finish(207, data=data, headers=headers)
+        return self.api_response(207, data=data, headers=headers)
 
     def _bad_request(self, error):
-        self.api_finish(400, error=error)
+        return self.api_response(400, error=error)
 
     def _unauthorized(self, error):
-        self.api_finish(401, error=error)
+        return self.api_response(401, error=error)
 
     def _not_found(self, error='Resource not found'):
-        self.api_finish(404, error=error)
+        return self.api_response(404, error=error)
 
     def _method_not_allowed(self, error):
-        self.api_finish(405, error=error)
+        return self.api_response(405, error=error)
 
     def _conflict(self, error):
-        self.api_finish(409, error=error)
+        return self.api_response(409, error=error)
 
     def _internal_server_error(self, error='Internal Server Error'):
-        self.api_finish(500, error=error)
+        return self.api_response(500, error=error)
 
     def _not_implemented(self):
-        self.api_finish(501)
+        return self.api_response(501)
 
     @classmethod
     def _raise_bad_request_error(cls, error):
@@ -351,9 +429,9 @@ def _parse_date(cls, value, fmt='%Y-%m-%d'):
 class NotFoundHandler(BaseRequestHandler):
     """A class used for the API v2 404 page."""
 
-    def get(self, *args, **kwargs):
+    def http_get(self, *args, **kwargs):
         """Get."""
-        self.api_finish(status=404)
+        return self.api_response(status=404)
 
     @classmethod
     def create_app_handler(cls, base):

medusa/server/api/v2/config.py

@@ -150,7 +150,7 @@ class ConfigHandler(BaseRequestHandler):
         'postProcessing.naming.stripYear': BooleanField(app, 'NAMING_STRIP_YEAR')
     }
 
-    def get(self, identifier, path_param=None):
+    def http_get(self, identifier, path_param=None):
         """Query general configuration.
 
         :param identifier:
@@ -180,7 +180,7 @@ def get(self, identifier, path_param=None):
 
         return self._ok(data=config_data)
 
-    def patch(self, identifier, *args, **kwargs):
+    def http_patch(self, identifier, *args, **kwargs):
         """Patch general configuration."""
         if not identifier:
             return self._bad_request('Config identifier not specified')
@@ -224,7 +224,7 @@ def patch(self, identifier, *args, **kwargs):
         })
         msg.push()
 
-        self._ok(data=accepted)
+        return self._ok(data=accepted)
 
 
 class DataGenerator(object):

medusa/server/api/v2/episodes.py

@@ -38,7 +38,7 @@ class EpisodeHandler(BaseRequestHandler):
     #: allowed HTTP methods
     allowed_methods = ('GET', 'PATCH', )
 
-    def get(self, series_slug, episode_slug, path_param):
+    def http_get(self, series_slug, episode_slug, path_param):
         """Query episode information.
 
         :param series_slug: series slug. E.g.: tvdb1234
@@ -79,7 +79,7 @@ def get(self, series_slug, episode_slug, path_param):
 
         return self._ok(data=data)
 
-    def patch(self, series_slug, episode_slug=None, path_param=None):
+    def http_patch(self, series_slug, episode_slug=None, path_param=None):
         """Patch episode."""
         series_identifier = SeriesIdentifier.from_slug(series_slug)
         if not series_identifier:
@@ -105,7 +105,7 @@ def patch(self, series_slug, episode_slug=None, path_param=None):
 
         accepted = self._patch_episode(episode, data)
 
-        self._ok(data=accepted)
+        return self._ok(data=accepted)
 
     def _patch_multi(self, series, request_data):
         """Patch multiple episodes."""
@@ -126,7 +126,7 @@ def _patch_multi(self, series, request_data):
 
             statuses[slug] = {'status': 200}
 
-        self._multi_status(data=statuses)
+        return self._multi_status(data=statuses)
 
     @staticmethod
     def _patch_episode(episode, data):