complete post request handler

requirements.txt

@@ -8,4 +8,5 @@ werkzeug
 filetype
 sqlalchemy
 flask_sqlalchemy
-rstr
+rstr
+pike

src/app.py

@@ -4,8 +4,9 @@ import sentry_sdk
 from flask import Flask
 from flask_restful import Api
 from sentry_sdk.integrations.flask import FlaskIntegration
+import pika
 
-from config import SENTRY_DSN, RELEASE_ID, RELEASEMODE, PORT, DEBUG
+from config import *
 import db
 from resources import *
 
@@ -27,13 +28,19 @@ if SENTRY_DSN:
         environment=RELEASEMODE
     )
 
+
 app = Flask(__name__)
+app.config['SQLALCHEMY_DATABASE_URI'] = f"postgresql://{POSTGRES_USERNAME}:{POSTGRES_PASSWORD}@{POSTGRES_HOSTNAME}:5432/{POSTGRES_DB}"
 api = Api(app)
 db.init_app(app)
 
 with app.app_context():
     db.create_all()
 
+rabbitmq = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST))
+rabbitmq_channel = rabbitmq.channel()
+rabbitmq_channel.queue_declare(RABBITMQ_QUEUE)
+
 formatter = logging.Formatter(
     fmt="%(asctime)s - %(levelname)s - %(module)s - %(message)s"
 )

src/config.py

@@ -20,3 +20,13 @@ DEBUG = os.environ.get("INPUT_SERVICE_DEBUG", True)
 SENTRY_DSN = os.environ.get("SENTRY_DSN")
 RELEASE_ID = os.environ.get("RELEASE_ID")
 RELEASEMODE = os.environ.get("INPUT_SERVICE_RELEASEMODE")
+
+RABBITMQ_HOST = os.getenv("INPUT_RABBITMQ_HOSTNAME", "localhost")
+RABBITMQ_EXCHANGE = os.getenv("INPUT_RABBITMQ_EXCHANGE", "wave-extract")
+RABBITMQ_QUEUE = os.getenv("INPUT_RABBITMQ_QUEUE", "wave-extract")
+POSTGRES_HOSTNAME = os.getenv("INPUT_POSTGRES_HOSTNAME", "localhost")
+POSTGRES_USERNAME = os.getenv("INPUT_POSTGRES_USERNAME", "input-service")
+POSTGRES_PASSWORD = os.getenv("INPUT_POSTGRES_PASSWORD", "input-service")
+POSTGRES_DB = os.getenv("INPUT_POSTGRES_DB", "input-service")
+
+STORAGE_HOSTNAME = os.getenv("INPUT_STORAGE_HOSTNAME", "localhost:8042")

src/models.py

@@ -22,4 +22,4 @@ class SampleMetadata(db.Model):
     device_id = db.Column(db.Integer, nullable=False)
     device_date = db.Column(db.DateTime, nullable=False)
 
-    tag = db.Column(db.String(32), nullable=False)
+    tag = db.Column(db.String(32), nullable=False, unique=True)

src/resources.py

@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
+from app import rabbitmq_channel
 import logging
 import json
 import rstr
 from flask_restful import Resource, reqparse
 from werkzeug.datastructures import FileStorage
+import requests
 import filetype
 import db
 from models import SampleMetadata
 from schemas import *
+from config import *
 
 """
 Flask Restful endpoints
@@ -20,7 +23,6 @@ __version__text__ = "1"
 
 LOGGER = logging.getLogger(__name__)
 
-
 class SampleResource(Resource):
     """
     Sample endpoint
@@ -58,10 +60,29 @@ class SampleResource(Resource):
                     'message': 'Input JSON schema invalid'}, 417
         generated_tag = rstr.xeger(r'^[a-zA-Z]+[0-9a-zA-Z_]*$', 2, 32)
 
-        record = SampleMetadata(device_id=desc['device_id'], device_date=desc['date'], tag=generated_tag)
-        db.session.add(record)
+        record = SampleMetadata(
+            device_id=desc['device_id'],
+            device_date=desc['date'],
+            tag=generated_tag)
+        try:
+            db.session.add(record)
+            requests.post(
+                f"http://{STORAGE_HOSTNAME}/object",
+                files={
+                    'tag': (None, generated_tag),
+                    'file': (
+                        'wave.wav',
+                        soundFile,
+                        kind.mime)})
+            rabbitmq_channel.basic_publish(
+                exchange=RABBITMQ_EXCHANGE,
+                routing_key='feature',
+                body=generated_tag)
+        except Exception as e:
+            LOGGER.exception(e)
+            db.session.rollback()
+            return {"status": "exception", "message": str(
+                e), "hint": "DB or downstream service error"}, 569
+
         db.session.commit()
-
-
-# TODO: Pass data and tag to ObjectStorage
-# TODO: Pass tag to MQ
+        return {"status": "ok", "message": generated_tag}, 200