Reputation: 519
I am trying to conduct a database test where I run a transaction set against the database while having X number of idle connections open to the database. That being said, I have started to create some logic for opening connections to the database as such:
import socket
def open_connect(hostname, port):
"""Function that creates a socket connection
:hostname: Hostname to connect to.
:port: Port to connect to host on.
"""
ipaddr = socket.gethostbyname(hostname)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((ipaddr, port))
sock.close()
def main():
"""Main function."""
open_connect('somedb', 3306)
if __name__ == "__main__":
main()
What I need to add now is some logic that will keep the connection open during the duration of the test. Can someone please provide guidance on the best way to keep a db connection open while I run actual queries in parallel? This will meet my test objectives. Thank you in advance.
Upvotes: 0
Views: 418
Reputation: 676
I think you might not want to use Sockets. I would try using a library like MySQLdb.
import MySQLdb
# connect
con = MySQLdb.connect(host="localhost", user="appuser", passwd="",db="onco")
You would be able to check if the connections is up using the ping method:
def is_connection_up(connection):
if connection.ping() == 0:
return True
else:
return False
So you can do something like this:
import MySQLdb
import time
def is_connection_up(connection):
if connection.ping() == 0:
return True
else:
return False
host,user,pwd,db = "localhost","user","pwd","db"
connections = [MySQLdb.connect(host=host,user=user,passwd=pwd,db=db) for i in range(10)]
while True:
time.sleep(10)
for con in connections:
if is_connection_up(con) is not True:
raise Exception("Connection is down, test might have gone wrong")
Upvotes: 2