working on queue, finalization

Consumer.py

@@ -26,6 +26,22 @@ redis_port = os.environ["REDIS_PORT"]
 redis_password = os.environ["REDIS_PASSWORD"]
 
 
+##################################################
+# PROCESSING METHODS
+##################################################
+
+def redis_send(user_id,msg_id,answer):
+    r = redis.Redis(
+        host=redis_host,
+        port=redis_port,
+        decode_responses=True,
+        username="default",
+        password=redis_password,
+    )
+ 
+    print(f"STORING ANSWER IN REDIS FOR USER_ID {user_id} AND MSG_ID {msg_id}")
+    success = r.set(f'ANSWER_FOR_USER_ID{user_id}_OF_{msg_id}',json.dumps(answer))
+    return success
 
 def model_call(request,token):
     # fill with last state 
@@ -40,45 +56,47 @@ def model_call(request,token):
     print(f"MODEL CALL WITH STATE {state} and PROMPT {request['prompt']} and MEMORY {request['memory']} and HT_TOKEN {token} and USER_EMAIL {request['user_email']} and USER_NAME {request['user_name']}")
     answer = get_response(request['prompt'], request['memory'],token,state,request['user_email'],request['user_name'])
     
-    print(f"ANSWER {answer}")
     # drop unserlizable keys
     for k in ["llm","rag_model"]:
         answer[k] = ""
 
+    print(f"ANSWER {answer}")
+    return answer
 
-    r = redis.Redis(
-        host=redis_host,
-        port=redis_port,
-        decode_responses=True,
-        username="default",
-        password=redis_password,
-    )
-    
-    user_id = request["user_id"]
-    msg_id = request["msg_id"]
-    print(f"STORING ANSWER IN REDIS FOR USER_ID {user_id} AND MSG_ID {msg_id}")
-    success = r.set(f'ANSWER_FOR_USER_ID{user_id}_OF_{msg_id}',json.dumps(answer))
+def process_message(recieved_msg):
+    # decrypt token
+    token = decrypt_token_from_json(json.loads(recieved_msg['ht_token_encrypted_dumped']))
+    print(f"TYPE {type(recieved_msg)}, CONTENT {recieved_msg} , TOKEN {token}")
+    # call the model
+    model_answer = model_call(recieved_msg,token)
+    # send answer to redis
+    user_id = recieved_msg["user_id"]
+    msg_id = recieved_msg["msg_id"]
+    redis_send_res = redis_send(user_id,msg_id,model_answer)
+    print({"STATUS": redis_send_res , "user_id": user_id, "msg_id": msg_id})
+   
 
+##################################################
+# CONSUMER METHODS
+##################################################
 
-    return {"STATUS": success , "user_id": user_id, "msg_id": msg_id}
 
 def get_connection():
     params = pika.URLParameters(RABBITMQ_URL)
     return pika.BlockingConnection(params)
 
 def callback(ch, method, properties, body):
+    ##### Recieve message and process it
     recieved_msg = json.loads(body.decode())
     print("-------------------------------------------------")
     print(f"MSG AT CONSUMER {consumer_id}" )
 
-    # simulate processing
-    token = decrypt_token_from_json(json.loads(recieved_msg['ht_token_encrypted_dumped']))
-    print(f"TYPE {type(recieved_msg)}, CONTENT {recieved_msg} , TOKEN {token}")
-    model_call(recieved_msg,token)
-
-    # (put your logic here)
+    ##### Process Message
+    process_message(recieved_msg)
     print(f"CONSUMER {consumer_id}:::: Processing done")
 
+
+    ###### Finalize
     ch.basic_ack(delivery_tag=method.delivery_tag)
 
 def start_consumer():
@@ -99,6 +117,10 @@ def start_consumer():
     print("Waiting for messages...")
     channel.start_consuming()
 
+##################################################
+# MAIN
+##################################################
+
 if __name__ == "__main__":
     print(f"Starting New Consumer {consumer_id}...")
     start_consumer()

encryption_utils.py

@@ -18,7 +18,6 @@ def encrypt_token_to_json(token: str) -> dict:
 
     encrypted = aesgcm.encrypt(iv, token.encode(), None)
 
-    # في AESGCM في Python: التاج (tag) بيكون في آخر 16 بايت
     ciphertext = encrypted[:-16]
     tag = encrypted[-16:]