Analytics service persist predictions

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

infra/openshift-manifests/analytics-service/kustomization.yaml

@@ -8,3 +8,4 @@ resources:
   - postgresql-datasource.yaml
   - analytics-broker.yaml
   - analytics-service.yaml
+  - predictions-trigger.yaml

infra/openshift-manifests/analytics-service/predictions-trigger.yaml

@@ -0,0 +1,13 @@
+apiVersion: eventing.knative.dev/v1
+kind: Trigger
+metadata:
+  name: analytics-service
+  namespace: ai-demo
+spec:
+  broker: predictions-broker
+  subscriber:
+    ref:
+      apiVersion: serving.knative.dev/v1
+      kind: Service
+      name: analytics-service
+    uri: /predictions

services/analytics-service/README.md

@@ -11,7 +11,7 @@ This image is published at `quay.io/kevent-mesh/ai-demo-analytics-service:main`.
 ```shell
 kubectl get routes -n knative-serving-ingress # Get analytics service route URL
 analytics_service_route_host="analytics-service-ai-demo.apps.<omitted>"
-curl -X POST -H "Content-Type: application/json" http://${analytics_service_route_host}/feedbacks -d '{"feedback": 1, "upload_id": "xyz"}'
+curl -X POST -H "Content-Type: application/json" http://${analytics_service_route_host}/feedbacks -d '{"feedback": 1, "uploadId": "xyz"}'
 ```
 
 

services/analytics-service/app.py

@@ -17,6 +17,21 @@ def init_feedbacks_table():
     cursor.close()
 
 
+def init_predictions_table():
+    cursor = conn.cursor()
+    cursor.execute('CREATE TABLE IF NOT EXISTS predictions ('
+                   'id serial PRIMARY KEY,'
+                   'probability DECIMAL(1, 12) NOT NULL,'
+                   'upload_id VARCHAR(200) NOT NULL,'
+                   'x0 DECIMAL(1, 12) NOT NULL,'
+                   'x1 DECIMAL(1, 12) NOT NULL,'
+                   'y0 DECIMAL(1, 12) NOT NULL,'
+                   'y1 DECIMAL(1, 12) NOT NULL,'
+                   'created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP);')
+    conn.commit()
+    cursor.close()
+
+
 def handler(signal, frame):
     print('Gracefully shutting down')
     sys.exit(0)
@@ -31,6 +46,7 @@ def handler(signal, frame):
 )
 
 init_feedbacks_table()
+init_predictions_table()
 
 signal.signal(signal.SIGINT, handler)
 signal.signal(signal.SIGTERM, handler)
@@ -39,11 +55,11 @@ def handler(signal, frame):
 
 
 @app.route("/feedbacks", methods=["POST"])
-def receive_event():
+def receive_feedbacks():
     # request body looks like this:
     # {
     #   "feedback": <int>
-    #   "upload_id": <string>
+    #   "uploadId": <string>
     # }
 
     print(f"Received request: {request.json}")
@@ -53,13 +69,47 @@ def receive_event():
     if 'feedback' not in body:
         return f"'feedback' not in request", 400
 
-    if 'upload_id' not in body:
-        return f"'upload_id' not in request", 400
+    if 'uploadId' not in body:
+        return f"'uploadId' not in request", 400
 
     cur = conn.cursor()
 
     cur.execute('INSERT INTO feedbacks (feedback, upload_id) VALUES (%s, %s)',
-                (body['feedback'], body['upload_id']))
+                (body['feedback'], body['uploadId']))
+
+    conn.commit()
+    cur.close()
+
+    return "", 204
+
+
+@app.route("/predictions", methods=["POST"])
+def receive_predictions():
+    # request body looks like this:
+    # {
+    #     "uploadId": "deadbeef",
+    #     "probability": "0.9012353451",
+    #     "x0": "0.24543",
+    #     "x1": "0.356647",
+    #     "y0": "0.34543",
+    #     "y1": "0.556647"
+    # }
+
+    print(f"Received request: {request.json}")
+
+    body = request.json
+
+    if 'probability' not in body:
+        return f"'probability' not in request", 400
+
+    if 'uploadId' not in body:
+        return f"'uploadId' not in request", 400
+
+    cur = conn.cursor()
+
+    cur.execute(
+        'INSERT INTO predictions (probability, upload_id, x0, x1, y0, y1) VALUES (%s, %s, %s, %s, %s, %s)',
+        (body['probability'], body['uploadId'], body['x0'], body['x1'], body['y0'], body['y1']))
 
     conn.commit()
     cur.close()