Testing pymysql and SQLAlchemy connections to mysql

'''

 

wget https://s3.amazonaws.com/rds-downloads/rds-combined-ca-bundle.pem

 

##connect to server

#mysql -uroot -hXXX.eu-west-2.rds.amazonaws.com --ssl-ca ./rds-combined-ca-bundle.pem -p

 

(Passwords hard coded as the test database should be deleted after it is done)

 

create database tmp_test_db CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

## Check it is utf-8

use mysql;

SELECT schema_name, default_character_set_name FROM information_schema.SCHEMATA;

CREATE USER ssl_user IDENTIFIED BY 'dsghyGFtuFG65G6t4rewsds';

grant ALL on tmp_test_db.* TO ssl_user@'%' REQUIRE SSL;

CREATE USER user IDENTIFIED BY 'DSF435354fhgGTH5T';

grant ALL on tmp_test_db.* TO user@'%';

FLUSH PRIVILEGES;

select host, user, ssl_type from user;

 

SSL Test

#mysql -ussl_user -hXXX.eu-west-2.rds.amazonaws.com -p"dsghyGFtuFG65G6t4rewsds" --ssl-ca ./rds-combined-ca-bundle.pem

 

No SSL Test

#mysql -uuser -hXXX.eu-west-2.rds.amazonaws.com -p"DSF435354fhgGTH5T"

'''

 

print("Start of test of DB connections")

 

import pymysql

 

user="user"

passwd="DSF435354fhgGTH5T"

ssluser="ssl_user"

sslpasswd="dsghyGFtuFG65G6t4rewsds"

 

host="XXX.eu-west-2.rds.amazonaws.com"

db="tmp_test_db"

 

 

def testDB_Using_pymysql_directly(user, passwd, host, db):

  ssl = {'ca': '/home/robert/dbtestTmp/rds-combined-ca-bundle.pem'}

  query = "select * from tasks;"

  db = pymysql.connect(host=host,user=user,passwd=passwd,db=db, ssl=ssl)

  cur = db.cursor()

  cur.execute(query)

  db.commit()

  result = cur.fetchall()

  print(result)

  db.close()

 

print("pymysql Testing non-SSL")

testDB_Using_pymysql_directly(user, passwd, host, db)

print("pymysql Testing SSL")

testDB_Using_pymysql_directly(ssluser, sslpasswd, host, db)

 

from sqlalchemy import create_engine

 

def testDB_Using_SQLAlchemy_and_pymysql(user, passwd, host, db):

  connectionString = "mysql+pymysql://" + user + ":" + passwd + "@" + host + "/" + db

  connect_args = {

    "ssl": {'ca': '/home/robert/dbtestTmp/rds-combined-ca-bundle.pem'}

  }

  engine = create_engine(connectionString, pool_recycle=3600, pool_size=40, max_overflow=0, connect_args=connect_args)

  connection = engine.connect()

 

print("SQLAlchemy Testing non-SSL")

testDB_Using_SQLAlchemy_and_pymysql(user, passwd, host, db)

print("SQLAlchemy Testing SSL")

testDB_Using_SQLAlchemy_and_pymysql(ssluser, sslpasswd, host, db)

 

 

print("End of test")