Skip to content

Update delta.py #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jul 3, 2025
Merged

Update delta.py #11

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 48 additions & 96 deletions delta.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
from cryptography import __version__ as cryptography_version
from urllib.parse import urlparse
import requests
from getpass import getpass

#import matplotlib.pyplot as plt

import oracledb
Expand All @@ -19,29 +21,6 @@

query_times = []

# Oracle Database credentials
oracle_un = 'admin'
oracle_pw = 'your_password'
oracle_cs = '(description= (retry_count=20)(retry_delay=3)(address=(protocol=tcps)(port=1521)(host=adb.uk-london-1.oraclecloud.com))(connect_data=(service_name=m783q0lhgfda8ox_demoadb_high.adb.oraclecloud.com))(security=(ssl_server_dn_match=yes)))'

# PostgreSQL credentials
pgsql_un = 'postgres'
pgsql_pw = 'your_password'
pgsql_host = 'localhost'
pgsql_port = '5432'
pgsql_db = 'postgres'

# MySQL credentials
mysql_un = 'mysql'
mysql_pw = 'your_password'
mysql_host = 'localhost'
mysql_port = '3306'
mysql_db = 'mysql'


# URL for testing
test_url = 'https://www.google.com'

def calculate_p99_latency():
if len(query_times) > 0:
p99_latency = np.percentile(query_times, 99)
Expand All @@ -59,13 +38,11 @@ def calculate_p99_latency():
print("++++++++++++++++++++++")
print("Mean Latency: {:.2f} ms".format(mean_latency))
print("++++++++++++++++++++++")

else:
print("No queries were executed.")

def oracle_ping(interval, csvfile):

conn = oracledb.connect(user=oracle_un, password=oracle_pw, dsn=oracle_cs)
def oracle_ping(interval, csvfile, user, password, dsn):
conn = oracledb.connect(user=user, password=password, dsn=dsn)
cursor = conn.cursor()
cursor.execute("select sys_context('USERENV','SID'), sys_context('USERENV','INSTANCE') from dual")
sid, instance = cursor.fetchone()
Expand All @@ -86,34 +63,8 @@ def oracle_ping(interval, csvfile):
cursor.close()
conn.close()



def postgresql_ping(interval, csvfile):
conn = psycopg2.connect(host=pgsql_host, port=pgsql_port, dbname=pgsql_db, user=pgsql_un, password=pgsql_pw)

cursor = conn.cursor()

t0 = time.perf_counter()
cursor.execute("SELECT 1")
cursor.fetchall()
t1 = time.perf_counter()

query_time = (t1 - t0) * 1000
query_times.append(query_time)

if csvfile is not None:
writer = csv.writer(csvfile)
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
writer.writerow([timestamp, query_time])

cursor.close()
conn.close()



def mysql_ping(interval, csvfile):
conn = pymysql.connect(host=mysql_host, port=int(mysql_port), user=mysql_un, password=mysql_pw, db=mysql_db)

def postgresql_ping(interval, csvfile, user, password, host, port, db):
conn = psycopg2.connect(host=host, port=port, dbname=db, user=user, password=password)
cursor = conn.cursor()

t0 = time.perf_counter()
Expand All @@ -132,12 +83,8 @@ def mysql_ping(interval, csvfile):
cursor.close()
conn.close()



def sql_server_ping(interval, csvfile):
conn_str = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={sql_server_host},{sql_server_port};DATABASE={sql_server_db};UID={sql_server_un};PWD={sql_server_pw}'
conn = pyodbc.connect(conn_str)

def mysql_ping(interval, csvfile, user, password, host, port, db):
conn = pymysql.connect(host=host, port=int(port), user=user, password=password, db=db)
cursor = conn.cursor()

t0 = time.perf_counter()
Expand All @@ -156,15 +103,11 @@ def sql_server_ping(interval, csvfile):
cursor.close()
conn.close()



def url_ping(interval, csvfile):
def url_ping(interval, csvfile, url):
t0 = time.perf_counter()
response = requests.get(test_url)
response = requests.get(url)
t1 = time.perf_counter()

# (The rest of the function remains the same)

query_time = (t1 - t0) * 1000
query_times.append(query_time)

Expand All @@ -173,54 +116,63 @@ def url_ping(interval, csvfile):
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
writer.writerow([timestamp, query_time])



# cmd-line arguements
parser = argparse.ArgumentParser(description="Connect and run a query.")
parser.add_argument("--interval", type=float, help="interval between each query, default 1", default=1)
parser.add_argument("--period", type=int, help="runtime in seconds; default 60", default=60)
parser.add_argument("--csvoutput", help="write timings to the named CSV file")
parser.add_argument("--db", choices=['oracle', 'postgresql', 'mysql', 'sqlserver', 'url'], required=True, help="specify the database or url to test")
parser.add_argument("--db", choices=['oracle', 'postgresql', 'mysql', 'url'], required=True, help="specify the database or url to test")
parser.add_argument("--user", help="Database username")
parser.add_argument("--password", help="Database password")
parser.add_argument("--host", help="Database host/DSN string")
parser.add_argument("--port", help="Database port")
parser.add_argument("--database", help="Database name")
parser.add_argument("--url", help="Test URL for latency")
args = parser.parse_args()


if args.csvoutput is not None:
csvfile = open(args.csvoutput, "w", newline="")
writer = csv.writer(csvfile)
writer.writerow(["Timestamp", "Query time (ms)", "SID", "Instance"])
if args.db == "oracle":
writer.writerow(["Timestamp", "Query time (ms)", "SID", "Instance"])
else:
writer.writerow(["Timestamp", "Query time (ms)"])
else:
csvfile = None


start_time = time.perf_counter()
end_time = start_time + args.period

# Main loop
while time.perf_counter() < end_time:
# Gather credentials if not provided
def prompt_if_none(val, prompt_text, secure=False):
if val:
return val
if secure:
return getpass(prompt_text)
return input(prompt_text)

for _ in range(int(args.period // args.interval)):
if args.db == 'oracle':
oracle_ping(args.interval, csvfile)
user = prompt_if_none(args.user, "Oracle Username: ")
password = prompt_if_none(args.password, "Oracle Password: ", secure=True)
dsn = prompt_if_none(args.host, "Oracle DSN (connection string): ")
oracle_ping(args.interval, csvfile, user, password, dsn)
elif args.db == 'postgresql':
postgresql_ping(args.interval, csvfile)
user = prompt_if_none(args.user, "Postgres Username: ")
password = prompt_if_none(args.password, "Postgres Password: ", secure=True)
host = prompt_if_none(args.host, "Postgres Host: ")
port = prompt_if_none(args.port, "Postgres Port: ")
db = prompt_if_none(args.database, "Postgres DB Name: ")
postgresql_ping(args.interval, csvfile, user, password, host, port, db)
elif args.db == 'mysql':
mysql_ping(args.interval, csvfile)
elif args.db == 'sqlserver':
sql_server_ping(args.interval, csvfile)
user = prompt_if_none(args.user, "MySQL Username: ")
password = prompt_if_none(args.password, "MySQL Password: ", secure=True)
host = prompt_if_none(args.host, "MySQL Host: ")
port = prompt_if_none(args.port, "MySQL Port: ")
db = prompt_if_none(args.database, "MySQL DB Name: ")
mysql_ping(args.interval, csvfile, user, password, host, port, db)
elif args.db == 'url':
url_ping(args.interval, csvfile)
url = prompt_if_none(args.url, "Test URL: ")
url_ping(args.interval, csvfile, url)
time.sleep(args.interval)

calculate_p99_latency()

# Plot the latencies on a graph
#plt.figure(figsize=(10, 5))
#plt.plot(query_times, marker='o')
#plt.title('Latency Over Time')
#plt.xlabel('Query Number')
#plt.ylabel('Latency (ms)')

# Save the plot to a file
output_file = "latency_plot.png"
#plt.savefig(output_file, bbox_inches='tight', dpi=300)

# Display the plot
#plt.show()