Initial implementation of background processing via RQ. Closes #4.

diffy/plugins/diffy_aws/plugin.py

@@ -9,10 +9,11 @@
 from typing import List
 
 import boto3
-from marshmallow import fields, Schema
+from marshmallow import fields
 
 from diffy.config import CONFIG
 from diffy.exceptions import TargetNotFound
+from diffy.schema import DiffyInputSchema
 from diffy.plugins import diffy_aws as aws
 from diffy.plugins.bases import PersistencePlugin, TargetPlugin, CollectionPlugin
 
@@ -30,7 +31,7 @@ def get_default_aws_account_number() -> dict:
     return sts.get_caller_identity()['Account']
 
 
-class AWSSchema(Schema):
+class AWSSchema(DiffyInputSchema):
     account_number = fields.String(default=get_default_aws_account_number, missing=get_default_aws_account_number)
     region = fields.String(default=CONFIG['DIFFY_DEFAULT_REGION'], missing=CONFIG['DIFFY_DEFAULT_REGION'])
 

diffy/schema.py

@@ -5,13 +5,89 @@
     :license: Apache, see LICENSE for more details.
 .. moduleauthor:: Kevin Glisson <[email protected]>
 """
-from marshmallow import fields, Schema, post_load
+
+from inflection import underscore, camelize
+from marshmallow import fields, Schema, post_load, pre_load, post_dump
 from marshmallow.exceptions import ValidationError
 
 from diffy.config import CONFIG
 from diffy.plugins.base import plugins
 
 
+class DiffySchema(Schema):
+    """
+    Base schema from which all diffy schema's inherit
+    """
+    __envelope__ = True
+
+    def under(self, data, many=None):
+        items = []
+        if many:
+            for i in data:
+                items.append(
+                    {underscore(key): value for key, value in i.items()}
+                )
+            return items
+        return {
+            underscore(key): value
+            for key, value in data.items()
+        }
+
+    def camel(self, data, many=None):
+        items = []
+        if many:
+            for i in data:
+                items.append(
+                    {camelize(key, uppercase_first_letter=False): value for key, value in i.items()}
+                )
+            return items
+        return {
+            camelize(key, uppercase_first_letter=False): value
+            for key, value in data.items()
+        }
+
+    def wrap_with_envelope(self, data, many):
+        if many:
+            if 'total' in self.context.keys():
+                return dict(total=self.context['total'], items=data)
+        return data
+
+
+class DiffyInputSchema(DiffySchema):
+    @pre_load(pass_many=True)
+    def preprocess(self, data, many):
+        return self.under(data, many=many)
+
+
+class DiffyOutputSchema(DiffySchema):
+    @pre_load(pass_many=True)
+    def preprocess(self, data, many):
+        if many:
+            data = self.unwrap_envelope(data, many)
+        return self.under(data, many=many)
+
+    def unwrap_envelope(self, data, many):
+        if many:
+            if data['items']:
+                self.context['total'] = data['total']
+            else:
+                self.context['total'] = 0
+                data = {'items': []}
+
+            return data['items']
+
+        return data
+
+    @post_dump(pass_many=True)
+    def post_process(self, data, many):
+        if data:
+            data = self.camel(data, many=many)
+        if self.__envelope__:
+            return self.wrap_with_envelope(data, many=many)
+        else:
+            return data
+
+
 def resolve_plugin_slug(slug):
     """Attempts to resolve plugin to slug."""
     plugin = plugins.get(slug)
@@ -26,7 +102,7 @@ class PluginOptionSchema(Schema):
     options = fields.Dict(missing={})
 
 
-class PluginSchema(Schema):
+class PluginSchema(DiffyInputSchema):
     options = fields.Dict(missing={})
 
     @post_load
@@ -54,3 +130,5 @@ class PayloadPluginSchema(PluginSchema):
 
 class AnalysisPluginSchema(PluginSchema):
     slug = fields.String(missing=CONFIG['DIFFY_ANALYSIS_PLUGIN'], default=CONFIG['DIFFY_ANALYSIS_PLUGIN'], required=True)
+
+

diffy_api/__init__.py

@@ -14,11 +14,12 @@
 from diffy_api import factory
 from diffy_api.baseline.views import mod as baseline_bp
 from diffy_api.analysis.views import mod as analysis_bp
-
+from diffy_api.tasks.views import mod as task_bp
 
 DIFFY_BLUEPRINTS = (
     baseline_bp,
-    analysis_bp
+    analysis_bp,
+    task_bp
 )
 
 

diffy_api/analysis/views.py

@@ -5,17 +5,17 @@
     :license: Apache, see LICENSE for more details.
 .. moduleauthor:: Kevin Glisson <[email protected]>
 """
-from flask import Blueprint, current_app
+from flask import Blueprint, current_app, request
 from flask_restful import reqparse, Api, Resource
 
-from diffy.core import analysis
 from diffy.plugins.base import plugins
 from diffy.exceptions import TargetNotFound
 
-from diffy_api.common.schema import validate_schema
+from diffy_api.core import async_analysis
+from diffy_api.common.util import validate_schema
 from diffy_api.schemas import (
     analysis_input_schema,
-    analysis_output_schema,
+    task_output_schema,
 )
 
 mod = Blueprint('analysis', __name__)
@@ -54,7 +54,7 @@ def get(self):
         data = plugins.get(current_app.config['DIFFY_PERSISTENCE_PLUGIN']).get_all('analysis')
         return data, 200
 
-    @validate_schema(analysis_input_schema, None)
+    @validate_schema(analysis_input_schema, task_output_schema)
     def post(self, data=None):
         """
         .. http:post:: /analysis
@@ -78,7 +78,7 @@ def post(self, data=None):
           :statuscode 403: unauthenticated
         """
         try:
-            return analysis(**data)
+            return async_analysis.queue(**request.json)
         except TargetNotFound as ex:
             return {'message': ex.message}, 404
 

diffy_api/baseline/views.py

@@ -5,16 +5,17 @@
     :license: Apache, see LICENSE for more details.
 .. moduleauthor:: Kevin Glisson <[email protected]>
 """
-from flask import Blueprint, current_app
+from flask import Blueprint, current_app, request
 from flask_restful import Api, Resource
 
-from diffy.core import baseline
 from diffy.plugins.base import plugins
 from diffy.exceptions import TargetNotFound
-from diffy_api.common.schema import validate_schema
+from diffy_api.core import async_baseline
+from diffy_api.common.util import validate_schema
 from diffy_api.schemas import (
     baseline_input_schema,
     baseline_output_schema,
+    task_output_schema,
 )
 
 
@@ -54,7 +55,7 @@ def get(self):
         data = plugins.get(current_app.config['DIFFY_PERSISTENCE_PLUGIN']).get_all('baseline')
         return data, 200
 
-    @validate_schema(baseline_input_schema, None)
+    @validate_schema(baseline_input_schema, task_output_schema)
     def post(self, data=None):
         """
         .. http:post:: /baselines
@@ -78,7 +79,7 @@ def post(self, data=None):
           :statuscode 403: unauthenticated
         """
         try:
-            return baseline(**data)
+            return async_baseline.queue(**request.json)
         except TargetNotFound as ex:
             return {'message': ex.message}, 404
 

diffy_api/common/schema.py → diffy_api/common/util.py

@@ -1,96 +1,12 @@
-"""
-.. module: diffy.common.schema
-    :platform: unix
-    :copyright: (c) 2018 by Netflix Inc., see AUTHORS for more
-    :license: Apache, see LICENSE for more details.
-
-.. moduleauthor:: Kevin Glisson <[email protected]>
-
-"""
-from functools import wraps
 from typing import List
-from flask import request, current_app
 
-from inflection import camelize, underscore
-from marshmallow import Schema, post_dump, pre_load
+from flask import request, current_app
+from functools import wraps
+from inflection import camelize
 
 from diffy_api.extensions import sentry
 
 
-class DiffySchema(Schema):
-    """
-    Base schema from which all grouper schema's inherit
-    """
-    __envelope__ = True
-
-    def under(self, data, many=None):
-        items = []
-        if many:
-            for i in data:
-                items.append(
-                    {underscore(key): value for key, value in i.items()}
-                )
-            return items
-        return {
-            underscore(key): value
-            for key, value in data.items()
-        }
-
-    def camel(self, data, many=None):
-        items = []
-        if many:
-            for i in data:
-                items.append(
-                    {camelize(key, uppercase_first_letter=False): value for key, value in i.items()}
-                )
-            return items
-        return {
-            camelize(key, uppercase_first_letter=False): value
-            for key, value in data.items()
-        }
-
-    def wrap_with_envelope(self, data, many):
-        if many:
-            if 'total' in self.context.keys():
-                return dict(total=self.context['total'], items=data)
-        return data
-
-
-class DiffyInputSchema(DiffySchema):
-    @pre_load(pass_many=True)
-    def preprocess(self, data, many):
-        return self.under(data, many=many)
-
-
-class DiffyOutputSchema(DiffySchema):
-    @pre_load(pass_many=True)
-    def preprocess(self, data, many):
-        if many:
-            data = self.unwrap_envelope(data, many)
-        return self.under(data, many=many)
-
-    def unwrap_envelope(self, data, many):
-        if many:
-            if data['items']:
-                self.context['total'] = data['total']
-            else:
-                self.context['total'] = 0
-                data = {'items': []}
-
-            return data['items']
-
-        return data
-
-    @post_dump(pass_many=True)
-    def post_process(self, data, many):
-        if data:
-            data = self.camel(data, many=many)
-        if self.__envelope__:
-            return self.wrap_with_envelope(data, many=many)
-        else:
-            return data
-
-
 def format_errors(messages: List[str]) -> dict:
     errors = {}
     for k, v in messages.items():
@@ -170,4 +86,4 @@ def decorated_function(*args, **kwargs):
             return unwrap_pagination(resp, output_schema), 200
 
         return decorated_function
-    return decorator
+    return decorator

diffy_api/core.py

@@ -0,0 +1,19 @@
+from diffy.core import baseline, analysis
+from diffy_api.extensions import rq
+from diffy_api.schemas import baseline_input_schema, analysis_input_schema
+
+
+@rq.job()
+def async_baseline(kwargs):
+    """Wrapper job around our standard baseline task."""
+    # we can't pickle our objects for remote works so we pickle the raw request and then load it here.
+    data = baseline_input_schema.load(kwargs)
+    return baseline(**data)
+
+
+@rq.job()
+def async_analysis(kwargs):
+    """Wrapper job around our standard analysis task."""
+    # we can't pickle our objects for remote works so we pickle the raw request and then load it here.
+    data = analysis_input_schema.load(kwargs)
+    return analysis(**data)

diffy_api/extensions.py

@@ -6,3 +6,6 @@
 """
 from raven.contrib.flask import Sentry
 sentry = Sentry()
+
+from flask_rq2 import RQ
+rq = RQ()

diffy_api/factory.py

@@ -18,7 +18,7 @@
 from diffy.common.utils import install_plugins
 
 from diffy_api.common.health import mod as health
-from diffy_api.extensions import sentry
+from diffy_api.extensions import sentry, rq
 
 
 DEFAULT_BLUEPRINTS = (
@@ -75,6 +75,7 @@ def configure_extensions(app):
     :param app:
     """
     sentry.init_app(app)
+    rq.init_app(app)
 
 
 def configure_blueprints(app, blueprints):