ssl_mysql_connect.py

import MySQLdb

def task():
    try:
        db = MySQLdb.connect(
            db="my_db", user=user, passwd=passwd,
            host=host,
            ssl={
                "key": "/etc/ssl/private/client-key.pem",
                "cert": "/etc/ssl/private/client-cert.pem",
            }
        )
        c = db.cursor()
    except Exception as e:
        print e
        return
    try:
        c.execute("select * from invalid_table")
    except Exception as e:
        print e
    finally:
        c.close()
        db.close()

import threading

ts = []
for i in range(10):
    t = threading.Thread(target=task)
    t.start()
    ts.append(t)

for t in ts:
    t.join()

ssl_mysql_connect_valid.py

import MySQLdb

def task():
    try:
        db = MySQLdb.connect(
            db="my_db", user=user, passwd=passwd,
            host=host,
            ssl={
                "key": "/etc/ssl/private/client-key.pem",
                "cert": "/etc/ssl/private/client-cert.pem",
            }
        )
        c = db.cursor()
    except Exception as e:
        print e
        return
    try:
        c.execute("select count(*) from example_table")
        print c.fetchall()
    except Exception as e:
        print e
    finally:
        c.close()
        db.close()

import threading

ts = []
for i in range(10):
    t = threading.Thread(target=task)
    t.start()
    ts.append(t)

for t in ts:
    t.join()

ssl_pymysql_connect_valid.py

import pymysql


def task():
    try:
        db = pymysql.connect(
            db="my_db", user=user, passwd=passwd,
            host=host,
            ssl={
                "key": "/etc/ssl/private/client-key.pem",
                "cert": "/etc/ssl/private/client-cert.pem",
            }
        )
        c = db.cursor()
    except Exception as e:
        print e
        return
    try:
        c.execute("select count(*) from example_table")
        print c.fetchall()
    except Exception as e:
        print e
    finally:
        c.close()
        db.close()


import threading


ts = []
for i in range(10):
    t = threading.Thread(target=task)
    t.daemon = True
    t.start()
    ts.append(t)

for t in ts:
    t.join()

ssl_sqlalchmey_connect_valid.py

import sqlalchemy


def task():
    try:
        engine = sqlalchemy.create_engine(
            'mysql://%s:%s@%s/' %
            (user, passwd, host),
            connect_args={
                "ssl": {
                   "key": "/etc/ssl/private/client-key.pem",
                   "cert": "/etc/ssl/private/client-cert.pem",
                }
            }
        )
        session = sqlalchemy.orm.sessionmaker(bind=engine)()
    except Exception as e:
        print e
        return
    try:
        print session.query(Model).count()
    except Exception as e:
        print e
    finally:
        session.close()


import threading


ts = []
for i in range(10):
    t = threading.Thread(target=task)
    t.daemon = True
    t.start()
    ts.append(t)

for t in ts:
    t.join()