Small code cleanups

requirements.txt

@@ -1,15 +1,15 @@
 sentry_sdk[flask]
-gunicorn
+gunicorn~=20.1.0
-Flask
+Flask~=2.0.1
-Flask-RESTful
+Flask-RESTful~=0.3.9
-requests
+requests~=2.26.0
 werkzeug
-sqlalchemy
+sqlalchemy~=1.4.22
-flask_sqlalchemy
+flask_sqlalchemy~=2.5.1
-xeger
+xeger~=0.3.5
-pika
+pika~=1.2.0
 psycopg2-binary
-marshmallow
+marshmallow~=3.13.0
-marshmallow-sqlalchemy
+marshmallow-sqlalchemy~=0.26.1
 flask-marshmallow
 py-healthcheck

src/app.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-import logging
 from flask import Flask
 from flask_restful import Api
 import sentry_sdk
@@ -7,14 +6,14 @@ from sentry_sdk.integrations.flask import FlaskIntegration
 from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
 from healthcheck import HealthCheck
 
-from config import *
+from config import Config
 from db import db
 from marshm import ma
 from resources import SampleResource, SampleParameterResource
 from healthchecks import health_database_status
 
 """
-Main Flask RESTful APIm
+Main Flask RESTful API
 """
 
 __author__ = "@tormakris"
@@ -22,34 +21,31 @@ __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "app"
 __version__text__ = "1"
 
-if SENTRY_DSN:
+if Config.SENTRY_DSN:
     sentry_sdk.init(
-        dsn=SENTRY_DSN,
+        dsn=Config.SENTRY_DSN,
         integrations=[FlaskIntegration(), SqlalchemyIntegration()],
         traces_sample_rate=1.0,
         send_default_pii=True,
-        release=RELEASE_ID,
+        release=Config.RELEASE_ID,
-        environment=RELEASEMODE,
+        environment=Config.RELEASEMODE,
         _experiments={"auto_enabling_integrations": True}
     )
 
 app = Flask(__name__)
-app.config[
+app.config.from_object(Config)
-    'SQLALCHEMY_DATABASE_URI'] = f"postgresql://{POSTGRES_USERNAME}:{POSTGRES_PASSWORD}@{POSTGRES_HOSTNAME}:5432/{POSTGRES_DB}?sslmode=require"
+
-app.config['EXCHANGE_NAME'] = RABBITMQ_EXCHANGE
-app.config['FLASK_PIKA_PARAMS'] = {'host': RABBITMQ_HOST,
-                                   'username': RABBITMQ_USERNAME,
-                                   'password': RABBITMQ_PASSWORD,
-                                   'port': 5672,
-                                   'virtual_host': '/'}
 api = Api(app)
 health = HealthCheck()
 db.init_app(app)
 ma.init_app(app)
 
-with app.app_context():
+
+@app.before_first_request
+def init_db():
     db.create_all()
 
+
 api.add_resource(SampleResource, "/sample")
 api.add_resource(SampleParameterResource, '/sample/<tag>')
 

src/config.py

@@ -1,35 +1,40 @@
 #!/usr/bin/env python3
 import os
 
-
 """
 Main Flask RESTful API
 """
 
-
 __author__ = "@tormakris"
 __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "app"
 __version__text__ = "1"
 
-
+_POSTGRES_HOSTNAME = os.getenv("INPUT_POSTGRES_HOSTNAME", "localhost")
-PORT = os.environ.get("INPUT_SERVICE_PORT", 8080)
+_POSTGRES_USERNAME = os.getenv("INPUT_POSTGRES_USERNAME", "input-service")
-DEBUG = os.environ.get("INPUT_SERVICE_DEBUG", True)
+_POSTGRES_PASSWORD = os.getenv("INPUT_POSTGRES_PASSWORD", "input-service")
+_POSTGRES_DB = os.getenv("INPUT_POSTGRES_DB", "input-service")
 
 
-SENTRY_DSN = os.environ.get("SENTRY_DSN")
+class Config:
-RELEASE_ID = os.environ.get("RELEASE_ID", "test")
+    PORT = int(os.environ.get("INPUT_SERVICE_PORT", 8080))
-RELEASEMODE = os.environ.get("INPUT_SERVICE_RELEASEMODE", "dev")
+    DEBUG = os.environ.get("INPUT_SERVICE_DEBUG", "true").lower() in ["true", "yes", "1"]
 
-RABBITMQ_HOST = os.getenv("INPUT_RABBITMQ_HOSTNAME", "localhost")
+    SENTRY_DSN = os.environ.get("SENTRY_DSN")
-RABBITMQ_EXCHANGE = os.getenv("INPUT_RABBITMQ_EXCHANGE", "dev")
+    RELEASE_ID = os.environ.get("RELEASE_ID", "test")
-RABBITMQ_QUEUE = os.getenv("INPUT_RABBITMQ_QUEUE", "wave-extract")
+    RELEASEMODE = os.environ.get("INPUT_SERVICE_RELEASEMODE", "dev")
-RABBITMQ_USERNAME = os.getenv("INPUT_RABBITMQ_USERNAME", "rabbitmq")
-RABBITMQ_PASSWORD = os.getenv("INPUT_RABBITMQ_PASSWORD", "rabbitmq")
 
-POSTGRES_HOSTNAME = os.getenv("INPUT_POSTGRES_HOSTNAME", "localhost")
+    EXCHANGE_NAME = os.getenv("INPUT_RABBITMQ_EXCHANGE", "dev")
-POSTGRES_USERNAME = os.getenv("INPUT_POSTGRES_USERNAME", "input-service")
+    RABBITMQ_QUEUE = os.getenv("INPUT_RABBITMQ_QUEUE", "wave-extract")
-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")
+    FLASK_PIKA_PARAMS = {
+        'host': os.getenv("INPUT_RABBITMQ_HOSTNAME", "localhost"),
+        'username': os.getenv("INPUT_RABBITMQ_USERNAME", "rabbitmq"),
+        'password': os.getenv("INPUT_RABBITMQ_PASSWORD", "rabbitmq"),
+        'port': int(os.getenv("INPUT_RABBITMQ_PORT", 5672)),
+        'virtual_host': '/'
+    }
+
+    SQLALCHEMY_DATABASE_URI = f"postgresql://{_POSTGRES_USERNAME}:{_POSTGRES_PASSWORD}@{_POSTGRES_HOSTNAME}:5432/{_POSTGRES_DB}?sslmode=require"
+
+    STORAGE_HOSTNAME = os.getenv("INPUT_STORAGE_HOSTNAME", "localhost:8042")

src/resources.py

@@ -2,13 +2,12 @@
 import json
 from xeger import Xeger
 from flask_restful import Resource
-from flask import request, current_app
+from flask import request, current_app, abort
 import requests
 import pika
 from db import db
 from models import SampleMetadata
 from schemas import SampleSchema, SampleMetadataSchema
-from config import *
 
 """
 Flask Restful endpoints
@@ -35,25 +34,23 @@ class SampleResource(Resource):
         :return:
         """
         if 'file' not in request.files:
-            return {"err_msg": "no file found"}, 469
+            return abort(400, "no file found")
         else:
             soundfile = request.files['file']
 
         if 'description' not in request.form:
-            return {"err_msg": "no description found"}, 470
+            return abort(400, "no description found")
         else:
             description = request.form.get("description")
 
         if soundfile.content_type != 'audio/wave':
-            current_app.logger.info(
+            current_app.logger.info(f"Input file was not WAV.")
-                f"Input file was not WAV.")
+            return abort(415, 'Input file not a wave file.')
-            return {'err_msg': 'Input file not a wave file.'}, 415
-
         try:
             desc = self.sampleschema.loads(description)
         except Exception as e:
             current_app.logger.exception(e)
-            return {'err_msg': 'Input JSON schema invalid'}, 417
+            return abort(417, 'Input JSON schema invalid')
 
         xeger = Xeger(limit=30)
         while True:
@@ -80,18 +77,26 @@ class SampleResource(Resource):
         record = SampleMetadata(
             device_id=desc['device_id'],
             device_date=desc['date'],
-            tag=generated_tag)
+            tag=generated_tag
-        try:
+        )
         db.session.add(record)
-            requests.post(
+
-                f"http://{STORAGE_HOSTNAME}/object",
+        files = {
-                files={
             'description': (None, json.dumps({'tag': generated_tag}), 'application/json'),
             'soundFile': (
                 'wave.wav',
                 soundfile,
                 soundfile.content_type,
-                        {'Content-Length': soundfile_content_length})}).raise_for_status()  # Anyádat curl am
+                {'Content-Length': soundfile_content_length})}
+
+        r = requests.post(
+            f"http://{current_app.config.get('STORAGE_HOSTNAME')}/object",
+            files=files)
+
+        if r.status_code not in [200, 201]:
+            return abort(500, f"Failed to upload sample to storage service. Upstream status: {r.status_code}: {r.text}")
+
+        try:
             credentials = pika.PlainCredentials(current_app.config['FLASK_PIKA_PARAMS']['username'],
                                                 current_app.config['FLASK_PIKA_PARAMS']['password'])
             connection = pika.BlockingConnection(
@@ -109,14 +114,13 @@ class SampleResource(Resource):
 
         except Exception as e:
             current_app.logger.exception(e)
-            db.session.rollback()
+            return abort(569, "AMPQ Publish error")
-            return {"err_msg": str(
-                e), "hint": "DB or downstream service error"}, 569
 
         db.session.commit()
         return {"tag": generated_tag}, 200
 
-    def get(self):
+
+def get(self):
     """
     Get all stored items
     :return:

src/schemas.py

@@ -34,5 +34,5 @@ class SampleMetadataSchema(ma.SQLAlchemyAutoSchema):
     """
     class Meta:
         model = SampleMetadata
-        exclude = ('timestamp', 'id', 'device_date',)
+        exclude = ('timestamp', 'id', 'device_date')
     date = auto_field("device_date", dump_only=False)