Implement database connection pooling with context manager pattern

- Added DBUtils PooledDB for intelligent connection pooling
- Created db_pool.py with lazy-initialized connection pool (max 20 connections)
- Added db_connection_context() context manager for safe connection handling
- Refactored all 19 database operations to use context manager pattern
- Ensures proper connection cleanup and exception handling
- Prevents connection exhaustion on POST requests
- Added logging configuration for debugging

Changes:
- py_app/app/db_pool.py: New connection pool manager
- py_app/app/logging_config.py: Centralized logging
- py_app/app/__init__.py: Updated to use connection pool
- py_app/app/routes.py: Refactored all DB operations to use context manager
- py_app/app/settings.py: Updated settings handlers
- py_app/requirements.txt: Added DBUtils dependency

This solves the connection timeout issues experienced with the fgscan page.
This commit is contained in:
Quality App System
2026-01-22 22:07:06 +02:00
parent fd801ab78d
commit 64b67b2979
9 changed files with 1928 additions and 920 deletions

View File

@@ -1,12 +1,37 @@
from flask import render_template, request, session, redirect, url_for, flash, current_app, jsonify
from .permissions import APP_PERMISSIONS, ROLE_HIERARCHY, ACTIONS, get_all_permissions, get_default_permissions_for_role
from .db_pool import get_db_connection
from .logging_config import get_logger
import mariadb
import os
import json
from contextlib import contextmanager
logger = get_logger('settings')
# Global permission cache to avoid repeated database queries
_permission_cache = {}
@contextmanager
def db_connection_context():
"""
Context manager for database connections from the pool.
Ensures connections are properly closed and committed/rolled back.
"""
logger.debug("Acquiring database connection from pool (settings)")
conn = get_db_connection()
try:
logger.debug("Database connection acquired successfully")
yield conn
except Exception as e:
logger.error(f"Error in settings database operation: {e}", exc_info=True)
conn.rollback()
raise e
finally:
if conn:
logger.debug("Closing database connection (settings)")
conn.close()
def check_permission(permission_key, user_role=None):
"""
Check if the current user (or specified role) has a specific permission.
@@ -18,40 +43,46 @@ def check_permission(permission_key, user_role=None):
Returns:
bool: True if user has the permission, False otherwise
"""
logger.debug(f"Checking permission '{permission_key}' for role '{user_role or session.get('role')}'")
if user_role is None:
user_role = session.get('role')
if not user_role:
logger.warning(f"Cannot check permission - no role provided")
return False
# Superadmin always has all permissions
if user_role == 'superadmin':
logger.debug(f"Superadmin bypass - permission '{permission_key}' granted")
return True
# Check cache first
cache_key = f"{user_role}:{permission_key}"
if cache_key in _permission_cache:
logger.debug(f"Permission '{permission_key}' found in cache: {_permission_cache[cache_key]}")
return _permission_cache[cache_key]
try:
conn = get_external_db_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT granted FROM role_permissions
WHERE role = %s AND permission_key = %s
""", (user_role, permission_key))
result = cursor.fetchone()
conn.close()
# Cache the result
has_permission = bool(result and result[0])
_permission_cache[cache_key] = has_permission
return has_permission
logger.debug(f"Checking permission '{permission_key}' for role '{user_role}' in database")
with db_connection_context() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT granted FROM role_permissions
WHERE role = %s AND permission_key = %s
""", (user_role, permission_key))
result = cursor.fetchone()
# Cache the result
has_permission = bool(result and result[0])
_permission_cache[cache_key] = has_permission
logger.info(f"Permission '{permission_key}' for role '{user_role}': {has_permission}")
return has_permission
except Exception as e:
print(f"Error checking permission {permission_key} for role {user_role}: {e}")
logger.error(f"Error checking permission {permission_key} for role {user_role}: {e}", exc_info=True)
return False
def clear_permission_cache():
@@ -226,31 +257,12 @@ def settings_handler():
# Helper function to get external database connection
def get_external_db_connection():
"""Reads the external_server.conf file and returns a MariaDB database connection."""
settings_file = os.path.join(current_app.instance_path, 'external_server.conf')
if not os.path.exists(settings_file):
raise FileNotFoundError("The external_server.conf file is missing in the instance folder.")
# Read settings from the configuration file
settings = {}
with open(settings_file, 'r') as f:
for line in f:
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith('#'):
continue
if '=' in line:
key, value = line.split('=', 1)
settings[key] = value
# Create a database connection
return mariadb.connect(
user=settings['username'],
password=settings['password'],
host=settings['server_domain'],
port=int(settings['port']),
database=settings['database_name']
)
"""
DEPRECATED: Use get_db_connection() from db_pool.py instead.
This function is kept for backward compatibility.
Returns a connection from the managed connection pool.
"""
return get_db_connection()
# User management handlers
def create_user_handler():