From 0313fbfee00696fd811240cc4cf64f861c678f41 Mon Sep 17 00:00:00 2001 From: George Date: Wed, 23 Sep 2026 00:40:11 +0530 Subject: [PATCH] feat: --ADDED connections matrix websocket api --- main.py | 186 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 186 insertions(+) diff --git a/main.py b/main.py index 2e72bb5..b0c49e2 100644 --- a/main.py +++ b/main.py @@ -397,11 +397,197 @@ async def database_status_monitor(websocket: WebSocket): print("Database status WebSocket disconnected") +#GET ACTIVE CONNECTIONS + +def get_active_connections( + data: DatabaseRequest +): + + conn = None + cursor = None + + try: + conn = psycopg2.connect( + host=data.host, + port=data.port, + dbname=data.dbname, + user=data.user, + password=data.password, + connect_timeout=3 + ) + + cursor = conn.cursor() + + cursor.execute(""" + SELECT COUNT(*) + FROM pg_stat_activity + WHERE state = 'active' + AND pid <> pg_backend_pid(); + """) + + active_connections = cursor.fetchone()[0] + + return { + "database_name": data.dbname, + "active_connections": active_connections + } + + except psycopg2.Error as e: + + raise HTTPException( + status_code=500, + detail=f"Failed to fetch active connections: {str(e)}" + ) + + finally: + if cursor: + cursor.close() + + if conn: + conn.close() +# HOW MANY CONNECTIONS ARE BEING USED +def get_connection_usage(data: DatabaseRequest): + conn = None + cursor = None + try: + conn = psycopg2.connect( + host=data.host, + port=data.port, + dbname=data.dbname, + user=data.user, + password=data.password, + connect_timeout=3 + ) + + cursor = conn.cursor() + + cursor.execute(""" + SELECT + COUNT(*) AS current_connections, + current_setting('max_connections')::int AS max_connections + FROM pg_stat_activity + GROUP BY current_setting('max_connections'); + """) + + row = cursor.fetchone() + + current_connections = row[0] + max_connections = row[1] + + connection_usage_percent = round( + (current_connections / max_connections) * 100, + 2 + ) + + return { + "current_connections": current_connections, + "max_connections": max_connections, + "connection_usage_percent": connection_usage_percent + } + + except psycopg2.Error as e: + + return { + "current_connections": None, + "max_connections": None, + "connection_usage_percent": None, + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + +# WEB SOCKETS FOR ACTIVE CONNECTIONS MONITORING +@app.websocket("/ws/database/active-connections") +async def database_active_connections_monitor(websocket: WebSocket): + + await websocket.accept() + + try: + + raw_data = await websocket.receive_json() + + data = DatabaseRequest(**raw_data) + + while True: + + try: + + ( + active_connections_result, + connection_usage_result + ) = await asyncio.wait_for( + + asyncio.gather( + + asyncio.to_thread( + get_active_connections, + data + ), + + asyncio.to_thread( + get_connection_usage, + data + ) + + ), + + timeout=5 + ) + + result = { + "database_name": data.dbname, + + "active_connections": + active_connections_result[ + "active_connections" + ], + + "current_connections": + connection_usage_result[ + "current_connections" + ], + + "max_connections": + connection_usage_result[ + "max_connections" + ], + + "connection_usage_percent": + connection_usage_result[ + "connection_usage_percent" + ] + } + + except asyncio.TimeoutError: + + result = { + "database_name": data.dbname, + "active_connections": None, + "current_connections": None, + "max_connections": None, + "connection_usage_percent": None + } + + await websocket.send_json(result) + + await asyncio.sleep(2) + + except WebSocketDisconnect: + + print( + "Database active connections WebSocket disconnected" + )