diving into adding connection to database ,simple test

app.py

@@ -2,9 +2,7 @@ from fastapi import FastAPI, HTTPException
 from typing import List, Union
 import requests
 import ast
-
-from sqlalchemy import create_engine, text
-from sqlalchemy.orm import sessionmaker, Session
+import pymysql
 
 # -------------------------
 # App Initialization
@@ -12,23 +10,25 @@ from sqlalchemy.orm import sessionmaker, Session
 app = FastAPI()
 
 # -------------------------
-# Database Setup (SQLAlchemy + PyMySQL)
+# Database Setup (Pure PyMySQL)
 # -------------------------
-DB_URL = (
-    "mysql+pymysql://avnadmin:AVNS_aT0RGFafs6_34WFegSF"
-    "@mysql-19285eb2-tahaelshrif1-7999.h.aivencloud.com:11520/trafficManagerSignals"
-    "?charset=utf8mb4"
-)
-
-engine = create_engine(DB_URL, pool_size=10, max_overflow=20, pool_pre_ping=True)
-SessionLocal = sessionmaker(bind=engine)
+timeout = 20
 
 def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
+    connection = pymysql.connect(
+        charset="utf8mb4",
+        connect_timeout=timeout,
+        cursorclass=pymysql.cursors.DictCursor,
+        db="trafficManagerSignals",
+        host="mysql-19285eb2-tahaelshrif1-7999.h.aivencloud.com",
+        password="AVNS_aT0RGFafs6_34WFegSF",
+        read_timeout=timeout,
+        port=11520,
+        user="avnadmin",
+        write_timeout=timeout,
+        ssl={"ssl": True},
+    )
+    return connection
 
 # -------------------------
 # Utilities
@@ -59,8 +59,6 @@ def get_traffic_in_container(coordinates):
 def read_root():
     return {"message": "Hello from FastAPI on Hugging Face Spaces!"}
 
-import requests
-
 @app.get("/test-internet")
 def test_internet():
     try:
@@ -68,13 +66,15 @@ def test_internet():
         return {"status": "ok", "your_ip": r.json()}
     except Exception as e:
         return {"status": "error", "detail": str(e)}
-    
+
 @app.get("/test-db")
 def test_db():
     try:
-        db = next(get_db())
-        
-        result = db.execute(text("SELECT 1")).fetchone()
+        connection = get_db()
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT 1")
+            result = cursor.fetchone()
+        connection.close()
         return {"status": "connected", "result": result}
     except Exception as e:
         return {"status": "error", "detail": str(e)}
@@ -95,22 +95,23 @@ def receive_coordinates(coords: Union[str, List[float]]):
 # -------------------------
 # Signal Database Check
 # -------------------------
-def check_signals(coords, db: Session = next(get_db())):
-    """
-    Check if lat/lon exists in traffic_signals DB table.
-    Returns list of tl_id_sumo.
-    """
+def check_signals(coords):
+    connection = get_db()
     found_signals = []
-    for lat, lon in coords:
-        result = db.execute(
-            text("""
-                SELECT tl_id_sumo FROM traffic_signals
-                WHERE lat = :lat AND lon = :lon
-            """),
-            {"lat": lat, "lon": lon}
-        ).fetchone()
-
-        if result:
-            found_signals.append(result.tl_id_sumo)
+    try:
+        with connection.cursor() as cursor:
+            for lat, lon in coords:
+                cursor.execute(
+                    """
+                    SELECT tl_id_sumo FROM traffic_signals
+                    WHERE lat = %s AND lon = %s
+                    """,
+                    (lat, lon)
+                )
+                result = cursor.fetchone()
+                if result:
+                    found_signals.append(result['tl_id_sumo'])
+    finally:
+        connection.close()
 
     return found_signals