pgbenchmark 0.1.7.1

A Python package to benchmark query performance on PostgreSQL Database.

pgbenchmark

Python package to benchmark query performance on a PostgreSQL database. It allows you to measure the execution time of queries over multiple runs, providing detailed metrics about each run's performance.

---

Installation

pip install pgbenchmark

---

Example

For ParallelBenchmark, scroll down....

import psycopg2
from pgbenchmark import Benchmark

conn = psycopg2.connect(
    dbname="postgres",
    user="postgres",
    password="  << Your Password >> ",
    host="localhost",
    port="5432"
)

benchmark = Benchmark(db_connection=conn, number_of_runs=1000)
benchmark.set_sql("SELECT 1;")

for result in benchmark:
    # {'run': X, 'sent_at': <DATETIME WITH MS>, 'duration': '0.000064'}
    pass

""" View Summary """
print(benchmark.get_execution_results())

# {'runs': 1000,
#      'min_time': '0.000576',
#      'max_time': '0.014741',
#      'avg_time': '0.0007',
#      'median_time': '0.000642',
#      'percentiles': {'p25': '0.000612',
#                      'p50': '0.000642',
#                      'p75': '0.000696',
#                      'p99': '0.001331'}
#      }

You can also pass SQL file, instead of query string

benchmark.set_sql("./test.sql")

---

Interactive | No-Code Mode

Simply run in your terminal:

pgbenchmark

You'll see the ouput

[ http://127.0.0.1:8000 ] Click to open pgbenchmark Interface

Configuration on the right, rest is very intuitive.

Pause and Resume buttons are not working for now :(

More Examples

Standard 'Benchmark' class allow all kinds of connections

1. Providing Nothing at all. Benchmark will use standard default factory values

from pgbenchmark import Benchmark

benchmark = Benchmark(number_of_runs=1000)
benchmark.set_sql("SELECT 1;")

for iteration in benchmark:
    pass

2. Providing Connection Details as Dict.

from pgbenchmark import Benchmark

params = {
    "dbname": "postgres",
    "host": "localhost",
    "port": "5432",
    "user": "postgres",
    "password": "postgres",
}

benchmark = Benchmark(db_connection=params, number_of_runs=1000)
benchmark.set_sql("SELECT 1;")

for iteration in benchmark:
    pass

3. Psycopg2 connection object directly

from pgbenchmark import Benchmark

params = {
    "dbname": "postgres",
    "host": "localhost",
    "port": "5432",
    "user": "postgres",
    "password": "postgres",
}

benchmark = Benchmark(db_connection=params, number_of_runs=1000)
benchmark.set_sql("SELECT 1;")

for iteration in benchmark:
    pass

---

Example with Parallel execution

⚠️ Please be careful. If you are running on Linux, pgbenchmark will load your cores on 100% !!!⚠️

from pgbenchmark import ParallelBenchmark  # <<-------- NEW IMPORT

conn_params = {
    "dbname": "postgres",
    "user": "postgres",
    "password": "",
    "host": "localhost",
    "port": "5432"
}

n_procs = 20  # Number of Processes (Cores basically)
n_runs_per_proc = 1_000

parallel_bench_pg = ParallelBenchmark(
    num_processes=n_procs,
    number_of_runs=n_runs_per_proc,
    db_connection_info=conn_params
)

parallel_bench_pg.set_sql("SELECT * from information_schema.tables;")  # Same as before

""" Unfortunately, as of now, you can't get execution results on the fly. """

parallel_bench_pg.run()  # RUN THE BENCHMARK 

results_pg = parallel_bench_pg.get_execution_results()
print(results_pg)

Example with Template Engine

From version 0.1.0 pgbenchmark supports simple Template Engine for queries.

import random
import string

from pgbenchmark import ParallelBenchmark

conn_params = {
    "dbname": "postgres",
    "user": "postgres",
    "password": "",
    "host": "localhost",
    "port": "5432"
}

n_procs = 20
n_runs_per_proc = 10


# Generator Function for Random Product Price
def generate_random_price():
    return round(random.randint(10, 1000), 2)


# Generator Function for Random Product Name (String)
def generate_random_string(length=10):
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for _ in range(length))


parallel_bench_pg = ParallelBenchmark(
    num_processes=n_procs,
    number_of_runs=n_runs_per_proc,
    db_connection_info=conn_params
)

# Define the SQL Query Template
query = """
            INSERT INTO products (name, price, stock_quantity) VALUES ('{{product_name}}', {{price_value}}, 10);
        """

# ===============================
# Note that similar to Jinja2, you have to define template variables within Query
#   {{product_name}}
#   {{price_value}}
# ===============================

parallel_bench_pg.set_sql(query)

# Set formatters
parallel_bench_pg.set_sql_formatter(for_placeholder="price_value", generator=generate_random_price)
parallel_bench_pg.set_sql_formatter(for_placeholder="product_name", generator=generate_random_string)

# Run Benchmark
if __name__ == '__main__':
    # Run the Parallel Benchmark
    parallel_bench_pg.run()

    results_pg = parallel_bench_pg.get_execution_results()

    throughput = results_pg["throughput_runs_per_sec"]
    avg_time = results_pg["avg_time"]

    print("\n=============================================================================")
    print("                           Benchmark Results                             ")
    print("=============================================================================")
    print(f"Throughput (runs/sec): {throughput}")
    print(f"Average Execution Time (sec): {avg_time}")

---