auto_trade_sys/test_db_pool.py
薇薇安 922a8f3820 1
2026-02-04 13:45:30 +08:00

43 lines
1.4 KiB
Python

import os
import sys
import logging
import traceback
import pymysql
from backend.database.connection import Database
# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
def test_connection():
print("Initializing Database...")
db = Database()
print("Testing get_connection...")
try:
with db.get_connection() as conn:
print(f"Connection type: {type(conn)}")
# Try to force DictCursor here if the fix in connection.py isn't enough,
# but we want to test if connection.py does it automatically.
with conn.cursor() as cursor:
cursor.execute("SELECT 1 as val")
result = cursor.fetchone()
print(f"Query result: {result}")
print(f"Result type: {type(result)}")
if isinstance(result, dict):
print("SUCCESS: Result is a dictionary")
else:
print("FAILURE: Result is NOT a dictionary")
# Debug: try to find how to set it
if hasattr(conn, 'connection'):
print(f"Inner connection type: {type(conn.connection)}")
except Exception as e:
print("Caught exception:")
traceback.print_exc()
if __name__ == "__main__":
test_connection()