server done

Author: JBris

.env

@@ -7,7 +7,7 @@ DB_NAME=iris
 DB_USER=user
 DB_PASSWORD=password
 DB_PORT=5432
-POSTGRES_DB_URL=postgresql://${DB_USER}:${DB_PASSWORD}@localhost/${DB_NAME}
+POSTGRES_DB_URL=postgresql://${DB_USER}:${DB_PASSWORD}@localhost:5432/${DB_NAME}
 
 # MetaBase
 METABASE_TAG=v0.48.0
@@ -20,10 +20,13 @@ CLOUDBEAVER_PORT=8978
 # Redis
 REDIS_TAG=7.2.2-bookworm
 REDIS_PORT=6379
+REDIS_INTERNAL_HOST=redis
+REDIS_INTERNAL_PORT=6379
 
 # ELK
 ELASTICSEARCH_TAG=8.11.0
 ELASTICSEARCH_PORT=9200
+ELASTICSEARCH_HOST=http://elasticsearch:9200
 
 # Adminer
 ADMINER_TAG=4.7.9-standalone

server/gql/resolvers.py

@@ -1,33 +1,63 @@
+from decouple import config
+import json
+from ml import get_predictions
 from model import models
 from model.database import DBSession
-from strawberry import ID
 from .schemas import Iris, PaginationInput
+from services import Cache
 from typing import List
 
+def get_data(pagination: PaginationInput) -> List[Iris]:
+	db = DBSession()
+		
+	try:
+		query = db.query(models.Iris)
+		if pagination is not None:
+			query = (
+				query
+				.offset(pagination.offset)
+				.limit(pagination.limit)
+			)
+		tasks = query.all()
+	
+	finally:
+		db.close()
+	
+	return tasks
+		
 class QueryResolver:
 	@staticmethod
 	def get_name() -> str:
 		return "Iris"
 
 	@staticmethod
 	def get_data(pagination: PaginationInput) -> List[Iris]:
-		db = DBSession()
-		
-		try:
-			query = db.query(models.Iris)
-			if pagination is not None:
-				query = (
-					query
-                    .offset(pagination.offset)
-                    .limit(pagination.limit)
-				)
-			tasks = query.all()
-    	
-		finally:
-			db.close()
-        
-		return tasks
-	
+		return get_data(pagination)
+
 	@staticmethod
 	def get_predictions(pagination: PaginationInput) -> List[Iris]:
-		return []
+		tasks = get_data(pagination)
+		
+		cache_host = config("REDIS_INTERNAL_HOST")
+		cache_port = config("REDIS_INTERNAL_PORT")
+		cache = Cache(
+			host=cache_host, port=cache_port
+		)
+
+		task_list = cache.get(cache.k)
+
+		if task_list is None:
+			task_list = []
+			for task in tasks:
+				task_dict = task.__dict__
+				del task_dict["_sa_instance_state"]
+				task_list.append(task_dict)
+
+			cache.set(
+				cache.k, json.dumps(task_list)
+			)
+		else:
+			task_list = json.loads(task_list)
+
+		pred_tasks = get_predictions(task_list, Iris)
+		return pred_tasks

server/gql/schemas.py

@@ -2,6 +2,7 @@
 
 @strawberry.type
 class Iris:
+    id: int
     sepal_length: float | None
     sepal_width: float | None
     petal_length: float | None
@@ -10,5 +11,5 @@ class Iris:
 
 @strawberry.input
 class PaginationInput:
-    offset: int = 50
-    limit: int = 100
+    offset: int = 0
+    limit: int = 250

server/ml/__init__.py

@@ -0,0 +1 @@
+from .logistic_regression import get_predictions

server/ml/logistic_regression.py

@@ -0,0 +1,48 @@
+from decouple import config
+from sklearn.model_selection import train_test_split
+import pandas as pd
+from services import Search
+from sklearn.linear_model import LogisticRegression
+from sklearn.metrics import (
+    accuracy_score, average_precision_score, f1_score, 
+    precision_score, recall_score
+)
+from sklearn.preprocessing import LabelEncoder
+
+estimator = LogisticRegression(
+    penalty = None, solver = "newton-cg", max_iter = 250, multi_class = "ovr"
+)
+
+encoder = LabelEncoder()
+
+def get_predictions(tasks, data_model):
+    df = pd.DataFrame(tasks).drop(columns=["id"])
+    resp = "species"
+    X = df.drop(columns=resp)
+    y = encoder.fit_transform(df[resp])
+    X_train, X_test, y_train, y_test = train_test_split(
+        X, y, test_size=0.2
+    )
+
+    estimator.fit(X_train, y_train)
+    preds = estimator.predict(X_test).reshape(-1, 1)
+    document = {
+        "accuracy": accuracy_score(y_test, preds),
+        "average_precision": average_precision_score(y_test, preds),
+        "f1": f1_score(y_test, preds, average="weighted"),
+        "precision": precision_score(y_test, preds, average="weighted"),
+        "recall": recall_score(y_test, preds, average="weighted"),
+    }
+
+    search_host = config("ELASTICSEARCH_HOST")
+    search = Search(search_host)
+    search.index(index=search.indx, document=document)
+    search.close()
+
+    preds = estimator.predict(X).reshape(-1, 1)
+    pred_tasks = []
+    for i, task_item in enumerate(tasks):
+        task_item[resp] = preds[i]
+        pred_task = data_model(**task_item)
+        pred_tasks.append(pred_task)
+    return pred_tasks

server/model/models.py

@@ -1,16 +1,11 @@
-from sqlalchemy import Column, String, Float
+from sqlalchemy import Column, String, Float, Integer
 from .database import Base
 
 class Iris(Base):
 	__tablename__ = "iris"
+	id = Column(Integer, primary_key = True)
 	sepal_length = Column(Float)
 	sepal_width = Column(Float)
 	petal_length = Column(Float)
 	petal_width = Column(Float)
 	species = Column(String)
-
-	def __repr__(self):
-		return 
-		f"""Iris(sepal_length={self.sepal_length}, 
-		sepal_width={self.sepal_width}, petal_length={self.petal_length},
-		petal_width={self.petal_width}, species={self.species})"""