Skip to content

Instantly share code, notes, and snippets.

@luqmansen
Last active August 18, 2022 16:55
  • Star 0 You must be signed in to star a gist
  • Fork 2 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save luqmansen/6c34b2a1700423b31602a4d4b1a02412 to your computer and use it in GitHub Desktop.
Simple Object Pool pattern to Create Connection Pool
from datetime import datetime
from multiprocessing import Lock
import mysql.connector
USER = "root"
PASSWD = "admin"
HOST = 'localhost'
DB = "db_test"
class Connection:
def __init__(self, user, pw, host, db):
self.created = datetime.now()
self.conn = mysql.connector.connect(user=user, password=pw, host=host, database=db)
def get_conn(self):
return self.conn
def close_conn(self):
return self.conn.close()
class ConnectionPool:
def __init__(self, conn_number):
self.conn_number = conn_number
self.pool = []
self.init_pool()
@staticmethod
def create_new_conn():
return Connection(USER, PASSWD, HOST, DB)
def init_pool(self):
for i in range(self.conn_number):
conn = self.create_new_conn()
self.pool.append(conn.get_conn())
def get_conn(self):
try:
with Lock():
return self.pool.pop()
except IndexError:
raise Exception("Connection Full")
def return_conn(self, conn: Connection):
with Lock():
self.pool.append(conn)
if __name__ == 'main':
from threading import Thread
pool = ConnectionPool(1)
def try_connection(pid):
print("Conn from pid ", pid)
try:
conn = pool.get_conn()
print(conn.get_server_info())
# try pulling connection without returning it
# pool.return_conn(conn)
except Exception as e:
print(str(e))
proc = []
for i in range(200):
proc.append(Thread(target=try_connection, args=(i,)))
[p.start() for p in proc]
[p.join() for p in proc]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment