benchmark.py

import time
from sqlalchemy import text
from attrs import define
import pyodbc
from mssql_python.connection import Connection


@define
class PerformanceTester():
    """Container for benchmark function"""

    @classmethod
    def benchmark_query(cls,
                        connection: pyodbc.Connection | Connection,
                        query: str,
                        driver_name: str,
                        iterations: int = 5) -> dict[str, str | float | int]:
        """Runs a query multiple times and returns timing statistics

        Args:
            connection (pyodbc.Connection | Connection): database connection
            query (str): select statement
            driver_name (str): driver name
            iterations (int, optional): number of times the query will be executed. Defaults to 5.

        Returns:
            dict[str, str | float | int]: results dictionary containing execution times.
        """
        times = []
        row_counts = []

        for i in range(iterations):
            start = time.perf_counter()

            if driver_name in ('pyodbc', 'pyodbc + SQLAlchemy'):
                cursor = connection.execute(text(query))
                result = cursor.fetchall()
                cursor.close()
            else:  # mssql-python
                cursor = connection.cursor()
                cursor.execute(query)
                result = cursor.fetchall()
                cursor.close()

            end = time.perf_counter()
            elapsed = end - start
            times.append(elapsed)
            row_counts.append(len(result))

        return {
            'driver': driver_name,
            'avg_time': sum(times) / len(times),
            'min_time': min(times),
            'max_time': max(times),
            'total_time': sum(times),
            'rows': row_counts[0],
            'iterations': iterations
        }