From bcc2b94a6aad4db996e44e19754310faa7c5f94e Mon Sep 17 00:00:00 2001 From: Dmitrii Gridnev Date: Tue, 4 Nov 2025 18:31:09 +0300 Subject: [PATCH] feat: add database profiler support for multiple databases - Introduced a new Database Profiler to track and log database operations for MongoDB, MySQL, PostgreSQL, SQLite, and Redis. - Updated `qase.config.json` to include a `profilers` array for enabling the database profiler. - Added comprehensive tests for each database type to ensure functionality. - Updated changelog to reflect the new version and changes. This release enhances the ability to profile database operations, providing detailed insights into query performance and execution metrics. --- examples/pytest/qase.config.json | 1 + examples/pytest/requirements.txt | 4 + .../pytest/tests/test_mongodb_profiler.py | 119 +++ examples/pytest/tests/test_mysql_profiler.py | 133 +++ .../pytest/tests/test_postgres_profiler.py | 133 +++ examples/pytest/tests/test_redis_profiler.py | 121 +++ examples/pytest/tests/test_sqlite_profiler.py | 109 ++ qase-python-commons/changelog.md | 6 + .../docs/DATABASE_PROFILERS.md | 193 ++++ qase-python-commons/pyproject.toml | 2 +- .../src/qase/commons/client/api_v2_client.py | 23 + .../src/qase/commons/models/step.py | 9 +- .../src/qase/commons/profilers/__init__.py | 7 +- .../src/qase/commons/profilers/db.py | 970 +++++++++++++++++- .../src/qase/commons/reporters/core.py | 5 +- 15 files changed, 1823 insertions(+), 12 deletions(-) create mode 100644 examples/pytest/tests/test_mongodb_profiler.py create mode 100644 examples/pytest/tests/test_mysql_profiler.py create mode 100644 examples/pytest/tests/test_postgres_profiler.py create mode 100644 examples/pytest/tests/test_redis_profiler.py create mode 100644 examples/pytest/tests/test_sqlite_profiler.py create mode 100644 qase-python-commons/docs/DATABASE_PROFILERS.md diff --git a/examples/pytest/qase.config.json b/examples/pytest/qase.config.json index e31f425b..e6548dd9 100644 --- a/examples/pytest/qase.config.json +++ b/examples/pytest/qase.config.json @@ -1,6 +1,7 @@ { "mode": "testops", "fallback": "report", + "profilers": ["db"], "report": { "driver": "local", "connection": { diff --git a/examples/pytest/requirements.txt b/examples/pytest/requirements.txt index 0b214b3e..53062f86 100644 --- a/examples/pytest/requirements.txt +++ b/examples/pytest/requirements.txt @@ -3,3 +3,7 @@ qase-pytest==6.0.0 qase-python-commons==3.0.2 qase-api-client==1.0.1 qase-api-v2-client==1.0.0 +psycopg2-binary>=2.9.0 +pymysql>=1.0.0 +pymongo>=4.0.0 +redis>=4.0.0 diff --git a/examples/pytest/tests/test_mongodb_profiler.py b/examples/pytest/tests/test_mongodb_profiler.py new file mode 100644 index 00000000..1a8760a2 --- /dev/null +++ b/examples/pytest/tests/test_mongodb_profiler.py @@ -0,0 +1,119 @@ +""" +Simple test for MongoDB database profiler +""" +import pytest +import os +from qase.pytest import qase + + +@qase.id(203) +@qase.title("MongoDB Database Profiler Test") +@qase.description("Simple test to verify MongoDB database profiler functionality") +@qase.severity("normal") +@qase.priority("high") +def test_mongodb_profiler(): + """Test MongoDB database operations with profiler.""" + try: + from pymongo import MongoClient + except ImportError: + pytest.skip("pymongo not installed") + + # Get connection parameters from environment variables or use defaults + db_host = os.getenv("MONGODB_HOST", "localhost") + db_port = int(os.getenv("MONGODB_PORT", "27017")) + db_name = os.getenv("MONGODB_DB", "testdb") + db_user = os.getenv("MONGODB_USER", None) + db_password = os.getenv("MONGODB_PASSWORD", None) + + client = None + + try: + # Connect to MongoDB + with qase.step("Connect to MongoDB database"): + if db_user and db_password: + client = MongoClient( + host=db_host, + port=db_port, + username=db_user, + password=db_password, + authSource="admin" + ) + else: + client = MongoClient( + host=db_host, + port=db_port + ) + + db = client[db_name] + collection = db.users + + # Insert data + with qase.step("Insert test users"): + result1 = collection.insert_one({ + "name": "John Doe", + "email": "john@example.com" + }) + result2 = collection.insert_one({ + "name": "Jane Smith", + "email": "jane@example.com" + }) + assert result1.inserted_id is not None + assert result2.inserted_id is not None + + # Find all documents + with qase.step("Query all users"): + users = list(collection.find()) + assert len(users) == 2, f"Expected 2 users, got {len(users)}" + + # Find specific document + with qase.step("Query user by name"): + user = collection.find_one({"name": "John Doe"}) + assert user is not None + assert user["email"] == "john@example.com" + + # Update document + with qase.step("Update user email"): + result = collection.update_one( + {"name": "John Doe"}, + {"$set": {"email": "john.doe@example.com"}} + ) + assert result.modified_count == 1 + + # Verify update + with qase.step("Query updated user"): + user = collection.find_one({"name": "John Doe"}) + assert user is not None + assert user["email"] == "john.doe@example.com" + + # Delete document + with qase.step("Delete user"): + result = collection.delete_one({"name": "Jane Smith"}) + assert result.deleted_count == 1 + + # Verify final state + with qase.step("Verify final user count"): + count = collection.count_documents({}) + assert count == 1, f"Expected 1 user, got {count}" + + except Exception as e: + # Check if it's a connection error + error_msg = str(e).lower() + if "connection" in error_msg or "timeout" in error_msg or "network" in error_msg: + pytest.skip(f"MongoDB not available: {e}") + else: + pytest.fail(f"Test failed with error: {e}") + finally: + # Clean up: drop collection and close connection + if client is not None: + try: + db = client[db_name] + collection = db.users + collection.drop() + except Exception: + pass + finally: + try: + client.close() + except Exception: + pass + diff --git a/examples/pytest/tests/test_mysql_profiler.py b/examples/pytest/tests/test_mysql_profiler.py new file mode 100644 index 00000000..df7f7716 --- /dev/null +++ b/examples/pytest/tests/test_mysql_profiler.py @@ -0,0 +1,133 @@ +""" +Simple test for MySQL database profiler +""" +import pytest +import os +from qase.pytest import qase + + +@qase.id(202) +@qase.title("MySQL Database Profiler Test") +@qase.description("Simple test to verify MySQL database profiler functionality") +@qase.severity("normal") +@qase.priority("high") +def test_mysql_profiler(): + """Test MySQL database operations with profiler.""" + try: + import pymysql + except ImportError: + pytest.skip("pymysql not installed") + + # Get connection parameters from environment variables or use defaults + db_host = os.getenv("MYSQL_HOST", "localhost") + db_port = int(os.getenv("MYSQL_PORT", "3306")) + db_name = os.getenv("MYSQL_DB", "testdb") + db_user = os.getenv("MYSQL_USER", "testuser") + db_password = os.getenv("MYSQL_PASSWORD", "testpass") + + conn = None + cursor = None + + try: + # Connect to MySQL + with qase.step("Connect to MySQL database"): + conn = pymysql.connect( + host=db_host, + port=db_port, + database=db_name, + user=db_user, + password=db_password + ) + + cursor = conn.cursor() + + # Create table + with qase.step("Create users table"): + cursor.execute(""" + CREATE TABLE IF NOT EXISTS users ( + id INT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + email VARCHAR(100) UNIQUE NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # Insert data + with qase.step("Insert test users"): + cursor.execute( + "INSERT INTO users (name, email) VALUES (%s, %s)", + ("John Doe", "john@example.com") + ) + cursor.execute( + "INSERT INTO users (name, email) VALUES (%s, %s)", + ("Jane Smith", "jane@example.com") + ) + + # Select data + with qase.step("Query all users"): + cursor.execute("SELECT * FROM users ORDER BY id") + users = cursor.fetchall() + assert len(users) == 2, f"Expected 2 users, got {len(users)}" + + # Update data + with qase.step("Update user email"): + cursor.execute( + "UPDATE users SET email = %s WHERE name = %s", + ("john.doe@example.com", "John Doe") + ) + assert cursor.rowcount == 1 + + # Select specific user + with qase.step("Query updated user"): + cursor.execute("SELECT * FROM users WHERE name = %s", ("John Doe",)) + user = cursor.fetchone() + assert user is not None + assert user[2] == "john.doe@example.com" + + # Delete data + with qase.step("Delete user"): + cursor.execute("DELETE FROM users WHERE name = %s", ("Jane Smith",)) + assert cursor.rowcount == 1 + + # Verify final state + with qase.step("Verify final user count"): + cursor.execute("SELECT COUNT(*) FROM users") + count = cursor.fetchone()[0] + assert count == 1, f"Expected 1 user, got {count}" + + # Commit transaction + conn.commit() + + except pymysql.err.OperationalError as e: + pytest.skip(f"MySQL not available: {e}") + except Exception as e: + pytest.fail(f"Test failed with error: {e}") + finally: + # Clean up: drop table and close connections + if cursor is not None: + try: + # Try to drop table only if connection is still valid + if conn is not None and conn.open: + try: + cursor.execute("DROP TABLE IF EXISTS users") + conn.commit() + except Exception: + # If drop fails, try to rollback + try: + conn.rollback() + except Exception: + pass + except Exception: + pass + finally: + try: + cursor.close() + except Exception: + pass + if conn is not None: + try: + if conn.open: + conn.close() + except Exception: + pass + diff --git a/examples/pytest/tests/test_postgres_profiler.py b/examples/pytest/tests/test_postgres_profiler.py new file mode 100644 index 00000000..c1bb3fbc --- /dev/null +++ b/examples/pytest/tests/test_postgres_profiler.py @@ -0,0 +1,133 @@ +""" +Simple test for PostgreSQL database profiler +""" +import pytest +import os +from qase.pytest import qase + + +@qase.id(200) +@qase.title("PostgreSQL Database Profiler Test") +@qase.description("Simple test to verify PostgreSQL database profiler functionality") +@qase.severity("normal") +@qase.priority("high") +def test_postgres_profiler(): + """Test PostgreSQL database operations with profiler.""" + try: + import psycopg2 + except ImportError: + pytest.skip("psycopg2 not installed") + + # Get connection parameters from environment variables or use defaults + db_host = os.getenv("POSTGRES_HOST", "localhost") + db_port = os.getenv("POSTGRES_PORT", "5432") + db_name = os.getenv("POSTGRES_DB", "testdb") + db_user = os.getenv("POSTGRES_USER", "testuser") + db_password = os.getenv("POSTGRES_PASSWORD", "testpass") + + conn = None + cursor = None + + try: + # Connect to PostgreSQL + with qase.step("Connect to PostgreSQL database"): + conn = psycopg2.connect( + host=db_host, + port=db_port, + database=db_name, + user=db_user, + password=db_password + ) + + cursor = conn.cursor() + + # Create table + with qase.step("Create users table"): + cursor.execute(""" + CREATE TABLE IF NOT EXISTS users ( + id SERIAL PRIMARY KEY, + name VARCHAR(100) NOT NULL, + email VARCHAR(100) UNIQUE NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # Insert data + with qase.step("Insert test users"): + cursor.execute( + "INSERT INTO users (name, email) VALUES (%s, %s)", + ("John Doe", "john@example.com") + ) + cursor.execute( + "INSERT INTO users (name, email) VALUES (%s, %s)", + ("Jane Smith", "jane@example.com") + ) + + # Select data + with qase.step("Query all users"): + cursor.execute("SELECT * FROM users ORDER BY id") + users = cursor.fetchall() + assert len(users) == 2, f"Expected 2 users, got {len(users)}" + + # Update data + with qase.step("Update user email"): + cursor.execute( + "UPDATE users SET email = %s WHERE name = %s", + ("john.doe@example.com", "John Doe") + ) + assert cursor.rowcount == 1 + + # Select specific user + with qase.step("Query updated user"): + cursor.execute("SELECT * FROM users WHERE name = %s", ("John Doe",)) + user = cursor.fetchone() + assert user is not None + assert user[2] == "john.doe@example.com" + + # Delete data + with qase.step("Delete user"): + cursor.execute("DELETE FROM users WHERE name = %s", ("Jane Smith",)) + assert cursor.rowcount == 1 + + # Verify final state + with qase.step("Verify final user count"): + cursor.execute("SELECT COUNT(*) FROM users") + count = cursor.fetchone()[0] + assert count == 1, f"Expected 1 user, got {count}" + + # Commit transaction + conn.commit() + + except psycopg2.OperationalError as e: + pytest.skip(f"PostgreSQL not available: {e}") + except Exception as e: + pytest.fail(f"Test failed with error: {e}") + finally: + # Clean up: drop table and close connections + if cursor is not None: + try: + # Try to drop table only if connection is still valid + if conn is not None and not conn.closed: + try: + cursor.execute("DROP TABLE IF EXISTS users") + conn.commit() + except Exception: + # If drop fails, try to rollback + try: + conn.rollback() + except Exception: + pass + except Exception: + pass + finally: + try: + cursor.close() + except Exception: + pass + if conn is not None: + try: + if not conn.closed: + conn.close() + except Exception: + pass + diff --git a/examples/pytest/tests/test_redis_profiler.py b/examples/pytest/tests/test_redis_profiler.py new file mode 100644 index 00000000..2ffe28b7 --- /dev/null +++ b/examples/pytest/tests/test_redis_profiler.py @@ -0,0 +1,121 @@ +""" +Simple test for Redis database profiler +""" +import pytest +import os +from qase.pytest import qase + + +@qase.id(204) +@qase.title("Redis Database Profiler Test") +@qase.description("Simple test to verify Redis database profiler functionality") +@qase.severity("normal") +@qase.priority("high") +def test_redis_profiler(): + """Test Redis database operations with profiler.""" + try: + import redis + except ImportError: + pytest.skip("redis not installed") + + # Get connection parameters from environment variables or use defaults + redis_host = os.getenv("REDIS_HOST", "localhost") + redis_port = int(os.getenv("REDIS_PORT", "6379")) + redis_password = os.getenv("REDIS_PASSWORD", None) + + r = None + + try: + # Connect to Redis + with qase.step("Connect to Redis database"): + if redis_password: + r = redis.Redis( + host=redis_host, + port=redis_port, + password=redis_password, + decode_responses=True + ) + else: + r = redis.Redis( + host=redis_host, + port=redis_port, + decode_responses=True + ) + + # Test connection + r.ping() + + # String operations + with qase.step("Set key-value pair"): + r.set("test_key", "test_value") + + with qase.step("Get key-value pair"): + value = r.get("test_key") + assert value == "test_value", f"Expected 'test_value', got '{value}'" + + with qase.step("Update key-value pair"): + r.set("test_key", "updated_value") + value = r.get("test_key") + assert value == "updated_value", f"Expected 'updated_value', got '{value}'" + + # List operations + with qase.step("List push operations"): + r.lpush("test_list", "item1", "item2", "item3") + + with qase.step("List range operation"): + items = r.lrange("test_list", 0, -1) + assert len(items) == 3, f"Expected 3 items, got {len(items)}" + + with qase.step("List pop operation"): + item = r.rpop("test_list") + assert item == "item1", f"Expected 'item1', got '{item}'" + + # Hash operations + with qase.step("Hash set operation"): + r.hset("test_hash", mapping={"field1": "value1", "field2": "value2"}) + + with qase.step("Hash get operation"): + value = r.hget("test_hash", "field1") + assert value == "value1", f"Expected 'value1', got '{value}'" + + with qase.step("Hash get all operation"): + all_fields = r.hgetall("test_hash") + assert len(all_fields) == 2, f"Expected 2 fields, got {len(all_fields)}" + + # Set operations + with qase.step("Set add operation"): + r.sadd("test_set", "member1", "member2", "member3") + + with qase.step("Set members operation"): + members = r.smembers("test_set") + assert len(members) == 3, f"Expected 3 members, got {len(members)}" + + # Delete operations + with qase.step("Delete key"): + deleted = r.delete("test_key") + assert deleted == 1 + + with qase.step("Verify key is deleted"): + value = r.get("test_key") + assert value is None, f"Expected None, got '{value}'" + + # Verify final state + with qase.step("Verify remaining keys"): + keys = r.keys("test_*") + assert len(keys) >= 3, f"Expected at least 3 test keys, got {len(keys)}" + + except redis.ConnectionError as e: + pytest.skip(f"Redis not available: {e}") + except Exception as e: + pytest.fail(f"Test failed with error: {e}") + finally: + # Clean up: delete all test keys + if r is not None: + try: + # Delete all test keys + test_keys = r.keys("test_*") + if test_keys: + r.delete(*test_keys) + except Exception: + pass + diff --git a/examples/pytest/tests/test_sqlite_profiler.py b/examples/pytest/tests/test_sqlite_profiler.py new file mode 100644 index 00000000..e0d47e16 --- /dev/null +++ b/examples/pytest/tests/test_sqlite_profiler.py @@ -0,0 +1,109 @@ +""" +Simple test for SQLite database profiler +""" +import pytest +import sqlite3 +import tempfile +import os +from qase.pytest import qase + + +@qase.id(201) +@qase.title("SQLite Database Profiler Test") +@qase.description("Simple test to verify SQLite database profiler functionality") +@qase.severity("normal") +@qase.priority("high") +def test_sqlite_profiler(): + """Test SQLite database operations with profiler.""" + # Create a temporary database file + with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as tmp: + db_path = tmp.name + + conn = None + cursor = None + + try: + # Connect to SQLite + with qase.step("Connect to SQLite database"): + conn = sqlite3.connect(db_path) + + cursor = conn.cursor() + + # Create table + with qase.step("Create users table"): + cursor.execute(""" + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + email TEXT UNIQUE NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # Insert data + with qase.step("Insert test users"): + cursor.execute( + "INSERT INTO users (name, email) VALUES (?, ?)", + ("John Doe", "john@example.com") + ) + cursor.execute( + "INSERT INTO users (name, email) VALUES (?, ?)", + ("Jane Smith", "jane@example.com") + ) + + # Select data + with qase.step("Query all users"): + cursor.execute("SELECT * FROM users ORDER BY id") + users = cursor.fetchall() + assert len(users) == 2, f"Expected 2 users, got {len(users)}" + + # Update data + with qase.step("Update user email"): + cursor.execute( + "UPDATE users SET email = ? WHERE name = ?", + ("john.doe@example.com", "John Doe") + ) + assert cursor.rowcount == 1 + + # Select specific user + with qase.step("Query updated user"): + cursor.execute("SELECT * FROM users WHERE name = ?", ("John Doe",)) + user = cursor.fetchone() + assert user is not None + assert user[2] == "john.doe@example.com" + + # Delete data + with qase.step("Delete user"): + cursor.execute("DELETE FROM users WHERE name = ?", ("Jane Smith",)) + assert cursor.rowcount == 1 + + # Verify final state + with qase.step("Verify final user count"): + cursor.execute("SELECT COUNT(*) FROM users") + count = cursor.fetchone()[0] + assert count == 1, f"Expected 1 user, got {count}" + + # Commit transaction + conn.commit() + + except Exception as e: + pytest.fail(f"Test failed with error: {e}") + finally: + # Clean up: close connections and remove database file + if cursor is not None: + try: + cursor.close() + except Exception: + pass + if conn is not None: + try: + conn.close() + except Exception: + pass + # Remove temporary database file + if os.path.exists(db_path): + try: + os.unlink(db_path) + except Exception: + pass + diff --git a/qase-python-commons/changelog.md b/qase-python-commons/changelog.md index 39775dad..d4fb4161 100644 --- a/qase-python-commons/changelog.md +++ b/qase-python-commons/changelog.md @@ -1,3 +1,9 @@ +# qase-python-commons@4.1.3 + +## What's new + +- Added support for database profilers. You can now profile your database operations with Qase. More information about database profilers can be found in the [docs](docs/DATABASE_PROFILERS.md). + # qase-python-commons@4.1.2 ## What's new diff --git a/qase-python-commons/docs/DATABASE_PROFILERS.md b/qase-python-commons/docs/DATABASE_PROFILERS.md new file mode 100644 index 00000000..1f521947 --- /dev/null +++ b/qase-python-commons/docs/DATABASE_PROFILERS.md @@ -0,0 +1,193 @@ +# Database Profiler + +## Overview + +The Database Profiler automatically tracks and logs all database operations during test execution. It captures database queries, execution times, and connection information, then sends this data as steps to Qase TestOps for detailed analysis and debugging. + +## Supported Databases + +The profiler supports the following database libraries: + +- **SQLite** - Built-in Python module (`sqlite3`) +- **PostgreSQL** - Via `psycopg2` or `psycopg2-binary` +- **MySQL** - Via `pymysql` +- **MongoDB** - Via `pymongo` +- **Redis** - Via `redis` (redis-py) + +## Configuration + +### Enable Database Profiler + +Add `"db"` to the `profilers` array in your `qase.config.json`: + +```json +{ + "profilers": ["db"] +} +``` + +### Multiple Profilers + +You can enable multiple profilers simultaneously: + +```json +{ + "profilers": ["db", "network", "sleep"] +} +``` + +## Collected Data + +For each database operation, the profiler collects the following information: + +### Query Information + +- **query** - The actual database query or operation (e.g., SQL statement, Redis command, MongoDB operation) +- **database_type** - Type of database (e.g., "PostgreSQL (psycopg2)", "MongoDB (pymongo)", "Redis") + +### Performance Metrics + +- **execution_time** - Time taken to execute the query (in seconds, with millisecond precision) +- **rows_affected** - Number of rows affected by the operation (when applicable) + +### Connection Information + +- **connection_info** - Database connection details (host, port, database name, etc.) + +### Error Information + +- **error** - Error message if the operation failed (only when `track_on_fail` is enabled) + +## Data Format in Qase TestOps + +When sent to Qase TestOps, database operations appear as test steps with the following structure: + +- **Action**: `[Database Type] query` (e.g., `[PostgreSQL (psycopg2)] SELECT * FROM users`) +- **Input Data**: Connection info, execution time, and rows affected formatted as: + + ``` + Connection: PostgreSQL: localhost | Execution time: 0.123s | Rows affected: 5 + ``` + +- **Status**: `passed` or `failed` (based on operation success) + +## Track on Fail + +By default, the profiler tracks database operations even when they fail. You can disable this behavior: + +```python +from qase.commons.profilers.db import DatabaseProfilerSingleton +from qase.commons.models.runtime import Runtime + +runtime = Runtime() +DatabaseProfilerSingleton.init(runtime=runtime, track_on_fail=False) +``` + +## Examples + +### SQL Query Example + +```python +import psycopg2 + +conn = psycopg2.connect(host="localhost", database="testdb", user="user", password="pass") +cursor = conn.cursor() +cursor.execute("SELECT * FROM users WHERE id = %s", (1,)) +users = cursor.fetchall() +``` + +**Collected Data:** + +- Query: `SELECT * FROM users WHERE id = %s` +- Database Type: `PostgreSQL (psycopg2)` +- Execution Time: `0.045s` +- Connection Info: `PostgreSQL: localhost` +- Rows Affected: `1` + +### MongoDB Operation Example + +```python +from pymongo import MongoClient + +client = MongoClient("localhost", 27017) +db = client.testdb +collection = db.users +collection.insert_one({"name": "John", "email": "john@example.com"}) +``` + +**Collected Data:** + +- Query: `insert_one({'name': 'John', 'email': 'john@example.com'})` +- Database Type: `MongoDB (pymongo)` +- Execution Time: `0.012s` +- Connection Info: `MongoDB: testdb.users` +- Rows Affected: `1` + +### Redis Command Example + +```python +import redis + +r = redis.Redis(host='localhost', port=6379, decode_responses=True) +r.set('key', 'value') +value = r.get('key') +``` + +**Collected Data:** + +- Query: `SET key value` +- Database Type: `Redis` +- Execution Time: `0.003s` +- Connection Info: `Redis: localhost:6379` + +## MongoDB Operations Tracked + +For MongoDB, the following operations are automatically tracked: + +- `find()` - Find multiple documents +- `find_one()` - Find a single document +- `insert_one()` - Insert a single document +- `update_one()` - Update a single document +- `delete_one()` - Delete a single document + +## Redis Operations Tracked + +For Redis, all commands executed through `execute_command()` are tracked, including: + +- String operations: `SET`, `GET`, `DEL`, etc. +- List operations: `LPUSH`, `RPOP`, `LRANGE`, etc. +- Hash operations: `HSET`, `HGET`, `HGETALL`, etc. +- Set operations: `SADD`, `SMEMBERS`, etc. +- And all other Redis commands + +## SQLAlchemy Support + +When using SQLAlchemy, all database operations are tracked automatically through SQLAlchemy's event system: + +```python +from sqlalchemy import create_engine, text + +engine = create_engine('postgresql://user:pass@localhost/db') +with engine.connect() as conn: + conn.execute(text("SELECT * FROM users")) +``` + +**Collected Data:** + +- Query: `SELECT * FROM users | params: None` +- Database Type: `SQLAlchemy` +- Execution Time: `0.056s` +- Connection Info: `SQLAlchemy Engine: postgresql://user:***@localhost/db` +- Rows Affected: `5` + +## Automatic Integration + +The database profiler works automatically once enabled in the configuration. No additional code changes are required - it intercepts database operations transparently using monkey patching and proxy classes. + +## Notes + +- The profiler only tracks operations that occur after it's enabled +- Failed operations are tracked by default (can be disabled with `track_on_fail=False`) +- The profiler handles errors gracefully and won't break your database operations +- All timing measurements have millisecond precision +- Connection information is extracted safely with fallbacks if extraction fails diff --git a/qase-python-commons/pyproject.toml b/qase-python-commons/pyproject.toml index 936c7354..d1f99d63 100644 --- a/qase-python-commons/pyproject.toml +++ b/qase-python-commons/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "qase-python-commons" -version = "4.1.2" +version = "4.1.3" description = "A library for Qase TestOps and Qase Report" readme = "README.md" authors = [{name = "Qase Team", email = "support@qase.io"}] diff --git a/qase-python-commons/src/qase/commons/client/api_v2_client.py b/qase-python-commons/src/qase/commons/client/api_v2_client.py index da34f98d..a2d7b4c7 100644 --- a/qase-python-commons/src/qase/commons/client/api_v2_client.py +++ b/qase-python-commons/src/qase/commons/client/api_v2_client.py @@ -155,6 +155,29 @@ def _prepare_step(self, project_code: str, step: Step) -> Dict: if step.step_type == StepType.SLEEP: prepared_step['data']['action'] = f"Sleep for {step.data.duration} seconds" + if step.step_type == StepType.DB_QUERY: + # Format database query as action + action_parts = [] + if step.data.database_type: + action_parts.append(f"[{step.data.database_type}]") + action_parts.append(step.data.query) + prepared_step['data']['action'] = " ".join(action_parts) + + # Add expected_result if available + if step.data.expected_result: + prepared_step['data']['expected_result'] = step.data.expected_result + + # Add connection info and execution time as input_data + info_parts = [] + if step.data.connection_info: + info_parts.append(f"Connection: {step.data.connection_info}") + if step.data.execution_time is not None: + info_parts.append(f"Execution time: {step.data.execution_time:.3f}s") + if step.data.rows_affected is not None: + info_parts.append(f"Rows affected: {step.data.rows_affected}") + if info_parts: + prepared_step['data']['input_data'] = " | ".join(info_parts) + if step.execution.attachments: uploaded_attachments = [] for file in step.execution.attachments: diff --git a/qase-python-commons/src/qase/commons/models/step.py b/qase-python-commons/src/qase/commons/models/step.py index 13e27f66..bd05b9b4 100644 --- a/qase-python-commons/src/qase/commons/models/step.py +++ b/qase-python-commons/src/qase/commons/models/step.py @@ -85,8 +85,15 @@ def add_response(self, status_code: int, response_body: Optional[str] = None, class StepDbQueryData(BaseModel): - def __init__(self, query: str, expected_result: str): + def __init__(self, query: str, expected_result: str = None, + database_type: str = None, execution_time: float = None, + rows_affected: int = None, connection_info: str = None): self.query = query + self.expected_result = expected_result + self.database_type = database_type + self.execution_time = execution_time + self.rows_affected = rows_affected + self.connection_info = connection_info class StepSleepData(BaseModel): diff --git a/qase-python-commons/src/qase/commons/profilers/__init__.py b/qase-python-commons/src/qase/commons/profilers/__init__.py index 1ccd7316..1d0ac6f5 100644 --- a/qase-python-commons/src/qase/commons/profilers/__init__.py +++ b/qase-python-commons/src/qase/commons/profilers/__init__.py @@ -1,10 +1,11 @@ from .network import NetworkProfiler, NetworkProfilerSingleton from .sleep import SleepProfiler -from .db import DbProfiler +from .db import DatabaseProfiler, DatabaseProfilerSingleton __all__ = [ NetworkProfiler, NetworkProfilerSingleton, SleepProfiler, - DbProfiler -] \ No newline at end of file + DatabaseProfiler, + DatabaseProfilerSingleton +] diff --git a/qase-python-commons/src/qase/commons/profilers/db.py b/qase-python-commons/src/qase/commons/profilers/db.py index ec328bb1..bf635ec2 100644 --- a/qase-python-commons/src/qase/commons/profilers/db.py +++ b/qase-python-commons/src/qase/commons/profilers/db.py @@ -1,8 +1,17 @@ +import sys +import time +import uuid +import threading +from functools import wraps +from typing import Optional, Any, Dict + from ..models.runtime import Runtime +from ..models.step import Step, StepDbQueryData, StepType -class DbProfiler: +class DatabaseProfiler: _instance = None + _lock = threading.Lock() def __init__(self, runtime: Runtime, track_on_fail: bool = True): self._original_functions = {} @@ -11,9 +20,960 @@ def __init__(self, runtime: Runtime, track_on_fail: bool = True): self.step = None def enable(self): - # TBD - return + """Enable database profiling for all supported database libraries.""" + # SQLAlchemy - try to enable (will skip if not available) + self._enable_sqlalchemy() + + # psycopg2 (PostgreSQL) - try to enable (will skip if not available) + self._enable_psycopg2() + + # pymysql (MySQL) - try to enable (will skip if not available) + self._enable_pymysql() + + # sqlite3 (built-in) - try to enable (will skip if not available) + self._enable_sqlite3() + + # pymongo (MongoDB) - try to enable (will skip if not available) + self._enable_pymongo() + + # redis-py - try to enable (will skip if not available) + self._enable_redis() def disable(self): - # TBD - return + """Disable database profiling and restore original functions.""" + for module_name, original_func in self._original_functions.items(): + if module_name == 'sqlalchemy': + # SQLAlchemy 2.0+ uses event listeners + try: + from sqlalchemy import event + from sqlalchemy.engine import Engine + + if isinstance(original_func, dict): + # Remove event listeners + if 'before' in original_func: + event.remove(Engine, "before_cursor_execute", original_func['before']) + if 'after' in original_func: + event.remove(Engine, "after_cursor_execute", original_func['after']) + except (ImportError, AttributeError): + pass + elif module_name == 'psycopg2': + import psycopg2 + psycopg2.connect = original_func + elif module_name == 'pymysql': + import pymysql.cursors + pymysql.cursors.Cursor.execute = original_func + elif module_name == 'sqlite3': + import sqlite3 + sqlite3.connect = original_func + elif module_name == 'pymongo': + import pymongo.collection + pymongo.collection.Collection.find = original_func.get('find') + pymongo.collection.Collection.find_one = original_func.get('find_one') + pymongo.collection.Collection.insert_one = original_func.get('insert_one') + pymongo.collection.Collection.update_one = original_func.get('update_one') + pymongo.collection.Collection.delete_one = original_func.get('delete_one') + elif module_name == 'redis': + import redis + redis.Redis.execute_command = original_func + + self._original_functions.clear() + + def _enable_sqlalchemy(self): + """Enable profiling for SQLAlchemy.""" + try: + import sqlalchemy + from sqlalchemy import event + from sqlalchemy.engine import Engine + + if 'sqlalchemy' not in self._original_functions: + # SQLAlchemy 2.0+ uses event listeners instead of monkey patching + # We'll use the before_cursor_execute and after_cursor_execute events + + def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany): + conn.info.setdefault('query_start_time', []).append(time.time()) + return statement, parameters + + def receive_after_cursor_execute(conn, cursor, statement, parameters, context, executemany): + try: + if conn.info.get('query_start_time'): + start_time = conn.info['query_start_time'].pop() + execution_time = time.time() - start_time + + query = str(statement) + if parameters: + try: + query += f" | params: {parameters}" + except: + pass + + DatabaseProfilerSingleton.get_instance()._log_db_query( + query=query, + database_type="SQLAlchemy", + execution_time=execution_time, + rows_affected=getattr(cursor, 'rowcount', None), + connection_info=f"SQLAlchemy Engine: {conn.engine.url}" + ) + except Exception: + pass # Don't break SQLAlchemy execution + + # Register event listeners + event.listen(Engine, "before_cursor_execute", receive_before_cursor_execute) + event.listen(Engine, "after_cursor_execute", receive_after_cursor_execute) + + # Store listeners for later removal + self._original_functions['sqlalchemy'] = { + 'before': receive_before_cursor_execute, + 'after': receive_after_cursor_execute + } + + except (ImportError, AttributeError): + pass + + def _enable_psycopg2(self): + """Enable profiling for psycopg2 (PostgreSQL).""" + try: + import psycopg2 + if 'psycopg2' not in self._original_functions: + # psycopg2.extensions.cursor is a C extension and cannot be monkey-patched directly + # Use monkey patching through connect function instead + self._original_functions['psycopg2'] = psycopg2.connect + psycopg2.connect = self._psycopg2_connect_wrapper(psycopg2.connect) + except ImportError: + pass + + def _enable_pymysql(self): + """Enable profiling for pymysql (MySQL).""" + try: + import pymysql.cursors + if 'pymysql' not in self._original_functions: + self._original_functions['pymysql'] = pymysql.cursors.Cursor.execute + pymysql.cursors.Cursor.execute = self._pymysql_execute_wrapper( + pymysql.cursors.Cursor.execute + ) + except ImportError: + pass + + def _enable_sqlite3(self): + """Enable profiling for sqlite3.""" + try: + import sqlite3 + if 'sqlite3' not in self._original_functions: + # SQLite3 методы нельзя переопределить напрямую + # Используем monkey patching через connect функцию + self._original_functions['sqlite3'] = sqlite3.connect + sqlite3.connect = self._sqlite3_connect_wrapper(sqlite3.connect) + except ImportError: + pass + + def _enable_pymongo(self): + """Enable profiling for pymongo (MongoDB).""" + try: + import pymongo.collection + if 'pymongo' not in self._original_functions: + self._original_functions['pymongo'] = { + 'find': pymongo.collection.Collection.find, + 'find_one': pymongo.collection.Collection.find_one, + 'insert_one': pymongo.collection.Collection.insert_one, + 'update_one': pymongo.collection.Collection.update_one, + 'delete_one': pymongo.collection.Collection.delete_one, + } + pymongo.collection.Collection.find = self._pymongo_find_wrapper( + pymongo.collection.Collection.find + ) + pymongo.collection.Collection.find_one = self._pymongo_find_one_wrapper( + pymongo.collection.Collection.find_one + ) + pymongo.collection.Collection.insert_one = self._pymongo_insert_wrapper( + pymongo.collection.Collection.insert_one + ) + pymongo.collection.Collection.update_one = self._pymongo_update_wrapper( + pymongo.collection.Collection.update_one + ) + pymongo.collection.Collection.delete_one = self._pymongo_delete_wrapper( + pymongo.collection.Collection.delete_one + ) + except ImportError: + pass + + def _enable_redis(self): + """Enable profiling for redis-py.""" + try: + import redis + if 'redis' not in self._original_functions: + self._original_functions['redis'] = redis.Redis.execute_command + redis.Redis.execute_command = self._redis_execute_wrapper( + redis.Redis.execute_command + ) + except ImportError: + pass + + def _sqlalchemy_execute_wrapper(self, func): + @wraps(func) + def wrapper(self, statement, *args, **kwargs): + start_time = time.time() + query = str(statement) if hasattr(statement, '__str__') else str(statement) + + try: + result = func(self, statement, *args, **kwargs) + execution_time = time.time() - start_time + + self._log_db_query( + query=query, + database_type="SQLAlchemy", + execution_time=execution_time, + rows_affected=getattr(result, 'rowcount', None), + connection_info=f"SQLAlchemy Engine: {self.url}" + ) + + return result + except Exception as e: + execution_time = time.time() - start_time + if self.track_on_fail: + self._log_db_query( + query=query, + database_type="SQLAlchemy", + execution_time=execution_time, + connection_info=f"SQLAlchemy Engine: {self.url}", + error=str(e) + ) + raise + + return wrapper + + def _psycopg2_connect_wrapper(self, func): + track_on_fail = self.track_on_fail + profiler_instance = self # Capture profiler instance + + class CursorProxy: + """Proxy class for psycopg2 cursor to intercept execute method.""" + def __init__(self, cursor, conn): + self._cursor = cursor + self._conn = conn + + def execute(self, query, *args, **kwargs): + """Execute query and log it.""" + start_time = time.time() + error_msg = None + + try: + result = self._cursor.execute(query, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info + try: + dsn_params = self._conn.get_dsn_parameters() + host = dsn_params.get('host', 'localhost') + except Exception: + host = 'localhost' + + # Get rowcount safely + try: + rows_affected = self._cursor.rowcount + except Exception: + rows_affected = None + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="PostgreSQL (psycopg2)", + execution_time=execution_time, + rows_affected=rows_affected, + connection_info=f"PostgreSQL: {host}" + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + error_msg = str(e) + execution_time = time.time() - start_time + + if track_on_fail: + try: + dsn_params = self._conn.get_dsn_parameters() + host = dsn_params.get('host', 'localhost') + except Exception: + host = 'localhost' + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="PostgreSQL (psycopg2)", + execution_time=execution_time, + connection_info=f"PostgreSQL: {host}", + error=error_msg + ) + except Exception: + # Silently ignore logging errors + pass + + # Re-raise the original exception + raise + + def __getattr__(self, name): + """Delegate all other attributes to the original cursor.""" + return getattr(self._cursor, name) + + class ConnectionProxy: + """Proxy class for psycopg2 connection to intercept cursor creation.""" + def __init__(self, conn): + self._conn = conn + + def cursor(self, *args, **kwargs): + """Create cursor and return proxy.""" + cursor = self._conn.cursor(*args, **kwargs) + return CursorProxy(cursor, self._conn) + + def __getattr__(self, name): + """Delegate all other attributes to the original connection.""" + return getattr(self._conn, name) + + @wraps(func) + def wrapper(*args, **kwargs): + # Get the original connection + conn = func(*args, **kwargs) + + # Return proxy instead of original connection + return ConnectionProxy(conn) + + return wrapper + + def _psycopg2_execute_wrapper(self, func): + @wraps(func) + def wrapper(self, query, *args, **kwargs): + start_time = time.time() + + try: + result = func(self, query, *args, **kwargs) + execution_time = time.time() - start_time + + self._log_db_query( + query=query, + database_type="PostgreSQL (psycopg2)", + execution_time=execution_time, + rows_affected=self.rowcount, + connection_info=f"PostgreSQL: {self.connection.get_dsn_parameters().get('host', 'localhost')}" + ) + + return result + except Exception as e: + execution_time = time.time() - start_time + if self.track_on_fail: + self._log_db_query( + query=query, + database_type="PostgreSQL (psycopg2)", + execution_time=execution_time, + connection_info=f"PostgreSQL: {self.connection.get_dsn_parameters().get('host', 'localhost')}", + error=str(e) + ) + raise + + return wrapper + + def _pymysql_execute_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, query, *args, **kwargs): + start_time = time.time() + error_msg = None + + try: + result = func(self, query, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MySQL: {self.connection.get_host_info()}" + except Exception: + connection_info = "MySQL" + + # Get rowcount safely + try: + rows_affected = self.rowcount + except Exception: + rows_affected = None + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MySQL (pymysql)", + execution_time=execution_time, + rows_affected=rows_affected, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + error_msg = str(e) + execution_time = time.time() - start_time + + if track_on_fail: + try: + connection_info = f"MySQL: {self.connection.get_host_info()}" + except Exception: + connection_info = "MySQL" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MySQL (pymysql)", + execution_time=execution_time, + connection_info=connection_info, + error=error_msg + ) + except Exception: + # Silently ignore logging errors + pass + + # Re-raise the original exception + raise + + return wrapper + + def _sqlite3_connect_wrapper(self, func): + track_on_fail = self.track_on_fail + + class CursorProxy: + """Proxy class for sqlite3 cursor to intercept execute method.""" + def __init__(self, cursor, conn): + self._cursor = cursor + self._conn = conn + + def execute(self, sql, *args, **kwargs): + """Execute query and log it.""" + start_time = time.time() + error_msg = None + + try: + result = self._cursor.execute(sql, *args, **kwargs) + execution_time = time.time() - start_time + + # Get rowcount safely + try: + rows_affected = self._cursor.rowcount + except Exception: + rows_affected = None + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=sql, + database_type="SQLite", + execution_time=execution_time, + rows_affected=rows_affected, + connection_info="SQLite" + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + error_msg = str(e) + execution_time = time.time() - start_time + + if track_on_fail: + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=sql, + database_type="SQLite", + execution_time=execution_time, + connection_info="SQLite", + error=error_msg + ) + except Exception: + # Silently ignore logging errors + pass + + # Re-raise the original exception + raise + + def __getattr__(self, name): + """Delegate all other attributes to the original cursor.""" + return getattr(self._cursor, name) + + class ConnectionProxy: + """Proxy class for sqlite3 connection to intercept cursor creation.""" + def __init__(self, conn): + self._conn = conn + + def cursor(self, *args, **kwargs): + """Create cursor and return proxy.""" + cursor = self._conn.cursor(*args, **kwargs) + return CursorProxy(cursor, self._conn) + + def __getattr__(self, name): + """Delegate all other attributes to the original connection.""" + return getattr(self._conn, name) + + @wraps(func) + def wrapper(*args, **kwargs): + # Get the original connection + conn = func(*args, **kwargs) + + # Return proxy instead of original connection + return ConnectionProxy(conn) + + return wrapper + + def _sqlite3_execute_wrapper(self, func): + @wraps(func) + def wrapper(self, sql, *args, **kwargs): + start_time = time.time() + + try: + result = func(self, sql, *args, **kwargs) + execution_time = time.time() - start_time + + self._log_db_query( + query=sql, + database_type="SQLite", + execution_time=execution_time, + rows_affected=self.rowcount, + connection_info=f"SQLite: {self.connection.execute('PRAGMA database_list').fetchone()}" + ) + + return result + except Exception as e: + execution_time = time.time() - start_time + if self.track_on_fail: + self._log_db_query( + query=sql, + database_type="SQLite", + execution_time=execution_time, + connection_info="SQLite", + error=str(e) + ) + raise + + return wrapper + + def _pymongo_find_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, filter=None, *args, **kwargs): + start_time = time.time() + query = f"find({filter})" + + try: + result = func(self, filter, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + raise + + return wrapper + + def _pymongo_find_one_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, filter=None, *args, **kwargs): + start_time = time.time() + query = f"find_one({filter})" + + try: + result = func(self, filter, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + rows_affected=1 if result is not None else 0, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + raise + + return wrapper + + def _pymongo_insert_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, document, *args, **kwargs): + start_time = time.time() + query = f"insert_one({document})" + + try: + result = func(self, document, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + rows_affected=1, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + raise + + return wrapper + + def _pymongo_update_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, filter, update, *args, **kwargs): + start_time = time.time() + query = f"update_one({filter}, {update})" + + try: + result = func(self, filter, update, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Get modified count safely + try: + rows_affected = result.modified_count + except Exception: + rows_affected = None + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + rows_affected=rows_affected, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + raise + + return wrapper + + def _pymongo_delete_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, filter, *args, **kwargs): + start_time = time.time() + query = f"delete_one({filter})" + + try: + result = func(self, filter, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Get deleted count safely + try: + rows_affected = result.deleted_count + except Exception: + rows_affected = None + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + rows_affected=rows_affected, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + connection_info = f"MongoDB: {self.database.name}.{self.name}" + except Exception: + connection_info = "MongoDB" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="MongoDB (pymongo)", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + raise + + return wrapper + + def _redis_execute_wrapper(self, func): + track_on_fail = self.track_on_fail + + @wraps(func) + def wrapper(self, command, *args, **kwargs): + start_time = time.time() + query = f"{command} {' '.join(map(str, args))}" + + try: + result = func(self, command, *args, **kwargs) + execution_time = time.time() - start_time + + # Get connection info safely + try: + host = self.connection_pool.connection_kwargs.get('host', 'localhost') + port = self.connection_pool.connection_kwargs.get('port', 6379) + connection_info = f"Redis: {host}:{port}" + except Exception: + connection_info = "Redis" + + # Log query - don't let logging break the execution + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="Redis", + execution_time=execution_time, + connection_info=connection_info + ) + except Exception: + # Silently ignore logging errors + pass + + return result + except Exception as e: + execution_time = time.time() - start_time + if track_on_fail: + try: + host = self.connection_pool.connection_kwargs.get('host', 'localhost') + port = self.connection_pool.connection_kwargs.get('port', 6379) + connection_info = f"Redis: {host}:{port}" + except Exception: + connection_info = "Redis" + + # Log error - don't let logging break the exception propagation + try: + profiler = DatabaseProfilerSingleton.get_instance() + profiler._log_db_query( + query=query, + database_type="Redis", + execution_time=execution_time, + connection_info=connection_info, + error=str(e) + ) + except Exception: + # Silently ignore logging errors + pass + + # Re-raise the original exception + raise + + return wrapper + + def _log_db_query(self, query: str, database_type: str, execution_time: float, + rows_affected: Optional[int] = None, connection_info: Optional[str] = None, + error: Optional[str] = None): + """Log database query as a step.""" + step_data = StepDbQueryData( + query=query, + database_type=database_type, + execution_time=execution_time, + rows_affected=rows_affected, + connection_info=connection_info + ) + + step = Step( + id=str(uuid.uuid4()), + step_type=StepType.DB_QUERY, + data=step_data + ) + + self.runtime.add_step(step) + + # Determine step status based on error + status = 'failed' if error else 'passed' + self.runtime.finish_step( + id=step.id, + status=status + ) + + +class DatabaseProfilerSingleton: + _instance = None + _lock = threading.Lock() + + @staticmethod + def init(**kwargs): + if DatabaseProfilerSingleton._instance is None: + with DatabaseProfilerSingleton._lock: + if DatabaseProfilerSingleton._instance is None: + DatabaseProfilerSingleton._instance = DatabaseProfiler(**kwargs) + + @staticmethod + def get_instance() -> DatabaseProfiler: + """Static access method""" + if DatabaseProfilerSingleton._instance is None: + raise Exception("Init plugin first") + return DatabaseProfilerSingleton._instance + + def __init__(self): + """Virtually private constructor""" + raise Exception("Use get_instance()") diff --git a/qase-python-commons/src/qase/commons/reporters/core.py b/qase-python-commons/src/qase/commons/reporters/core.py index 1ea72410..d96b7e66 100644 --- a/qase-python-commons/src/qase/commons/reporters/core.py +++ b/qase-python-commons/src/qase/commons/reporters/core.py @@ -135,8 +135,9 @@ def setup_profilers(self, runtime: Runtime) -> None: from ..profilers import SleepProfiler self.profilers.append(SleepProfiler(runtime=runtime)) if profiler == "db": - from ..profilers import DbProfiler - self.profilers.append(DbProfiler(runtime=runtime)) + from ..profilers import DatabaseProfilerSingleton + DatabaseProfilerSingleton.init(runtime=runtime) + self.profilers.append(DatabaseProfilerSingleton.get_instance()) def enable_profilers(self) -> None: if self.reporter: