Implement Redis integration and async database operations for anomaly results

app.py

@@ -1,30 +1,59 @@
+
 import time
 import os
-import requests
+import redis
+
+import asyncio
+from sql import insert_rca_result, connect_to_database, disconnect_from_database
 
+# Directories
 UPLOAD_DIR = "/data/uploads"
 PROCESSED_DIR = "/data/processed"
 os.makedirs(PROCESSED_DIR, exist_ok=True)
 
+# Redis configuration (must be set in environment variables)
+REDIS_HOST = os.environ["REDIS_HOST"]
+REDIS_PORT = int(os.environ["REDIS_PORT"])
+REDIS_QUEUE = os.environ["REDIS_QUEUE"]
+READY_FOR_RCA_QUEUE = os.environ.get("READY_FOR_RCA_QUEUE", "logbert_ready_for_rca")
+
+# Initialize Redis client
+redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
+
 def process_log(file_path):
     # Dummy anomaly detection logic; replace with your real pipeline
     # For example, call your detect_anomalies_and_explain here
     return {"filename": os.path.basename(file_path), "anomaly": True, "details": "Example anomaly detected."}
 
+
+
 def save_rca_to_db(rca_result):
-    # Replace with actual DB insert logic (e.g., via RCA API or direct DB connection)
-    requests.post("http://rca_api:8000/rca/", json=rca_result)
+    # Store anomaly result in rca_results table using sql.py
+    async def _save():
+        await connect_to_database()
+        await insert_rca_result(rca_result)
+        await disconnect_from_database()
+    asyncio.run(_save())
 
 def main():
-    processed_files = set()
     while True:
-        for filename in os.listdir(UPLOAD_DIR):
+        # Block until a filename is available in the Redis queue
+        filename = redis_client.rpop(REDIS_QUEUE)
+        if filename:
             file_path = os.path.join(UPLOAD_DIR, filename)
-            if filename not in processed_files and os.path.isfile(file_path):
+            if os.path.isfile(file_path):
                 rca_result = process_log(file_path)
                 save_rca_to_db(rca_result)
-                processed_files.add(filename)
-        time.sleep(5)
+                # Notify ready-for-rca queue
+                try:
+                    redis_client.lpush(READY_FOR_RCA_QUEUE, filename)
+                    print(f"Notified {READY_FOR_RCA_QUEUE} for {filename}")
+                except Exception as redis_exc:
+                    print(f"Failed to notify ready-for-rca queue: {redis_exc}")
+            else:
+                print(f"File not found: {file_path}")
+        else:
+            time.sleep(2)
 
 if __name__ == "__main__":
     main()

sql.py

@@ -0,0 +1,89 @@
+import os
+import logging
+from dotenv import load_dotenv
+import databases
+import sqlalchemy
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# Load the DATABASE_URL from environment variables
+
+# Explicitly load the .env file from the current project directory
+ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../.env")
+load_dotenv(dotenv_path=ENV_PATH)
+
+DATABASE_URL = os.getenv("DATABASE_URI")
+
+if not DATABASE_URL:
+    logger.error("DATABASE_URI not found in environment variables.")
+    raise ValueError("DATABASE_URI not found in environment variables. Ensure it is set in the .env file.")
+
+
+# Initialize the database connection and metadata
+try:
+    database = databases.Database(DATABASE_URL)
+    metadata = sqlalchemy.MetaData()
+    logger.info("Database connection initialized successfully.")
+except Exception as e:
+    logger.error("Failed to initialize database connection: %s", e)
+    raise
+
+# Define rca_results table
+rca_results = sqlalchemy.Table(
+    "rca_results",
+    metadata,
+    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
+    sqlalchemy.Column("filename", sqlalchemy.String, nullable=False),
+    sqlalchemy.Column("anomaly", sqlalchemy.Boolean, nullable=False),
+    sqlalchemy.Column("details", sqlalchemy.Text, nullable=True),
+)
+
+# Insert anomaly result into rca_results table
+async def insert_rca_result(rca_result):
+    query = rca_results.insert().values(
+        filename=rca_result["filename"],
+        anomaly=rca_result["anomaly"],
+        details=rca_result.get("details", "")
+    )
+    await database.execute(query)
+
+# Create async SQLAlchemy engine and session for PostgreSQL
+ASYNC_DATABASE_URL = DATABASE_URL.replace('postgresql://', 'postgresql+asyncpg://')
+engine = create_async_engine(ASYNC_DATABASE_URL, echo=False, future=True)
+async_session = sessionmaker(
+    engine, expire_on_commit=False, class_=AsyncSession
+)
+
+# Helper function to initialize database
+async def connect_to_database():
+    """
+    Connects to the database when the application starts.
+    """
+    try:
+        await database.connect()
+        logger.info("Successfully connected to the database.")
+    except Exception as e:
+        logger.error("Error connecting to the database: %s", e)
+        raise
+
+# Helper function to disconnect database
+async def disconnect_from_database():
+    """
+    Disconnects from the database when the application shuts down.
+    """
+    try:
+        await database.disconnect()
+        logger.info("Successfully disconnected from the database.")
+    except Exception as e:
+        logger.error("Error disconnecting from the database: %s", e)
+        raise