43 lines
889 B
Python
43 lines
889 B
Python
import mysql.connector
|
|
from mysql.connector import pooling
|
|
import threading
|
|
|
|
MYSQL_CONFIG = {
|
|
"host": "localhost",
|
|
"user": "florian",
|
|
"password": "password123++",
|
|
"database": "app"
|
|
}
|
|
|
|
# Lock to ensure thread-safe pool creation
|
|
_pool_lock = threading.Lock()
|
|
_connection_pool = None
|
|
|
|
def get_connection_pool():
|
|
global _connection_pool
|
|
with _pool_lock:
|
|
if _connection_pool is None:
|
|
_connection_pool = mysql.connector.pooling.MySQLConnectionPool(
|
|
pool_name="mypool",
|
|
pool_size=5,
|
|
pool_reset_session=True,
|
|
**MYSQL_CONFIG
|
|
)
|
|
return _connection_pool
|
|
|
|
# Dependency for FastAPI
|
|
def get_db():
|
|
pool = get_connection_pool()
|
|
conn = pool.get_connection()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
# Manual test
|
|
for conn in get_db():
|
|
cursor = conn.cursor(dictionary=True)
|
|
cursor.execute("SELECT NOW() AS ts")
|
|
print(cursor.fetchone())
|