86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
"""
|
|
Foundational SQL connection management for the Aether API.
|
|
Isolates the SQLAlchemy engine and global connection state to prevent circular imports.
|
|
"""
|
|
import logging
|
|
import threading
|
|
from typing import Any, Optional
|
|
from sqlalchemy import create_engine
|
|
from app.config import settings
|
|
|
|
log = logging.getLogger('root')
|
|
|
|
# 1. Thread-local storage for capturing last SQL error message
|
|
_sql_error_state = threading.local()
|
|
|
|
def get_last_sql_error() -> Optional[str]:
|
|
"""Retrieves and clears the last captured SQL error message."""
|
|
error = getattr(_sql_error_state, 'last_error', None)
|
|
_sql_error_state.last_error = None
|
|
return error
|
|
|
|
def set_last_sql_error(error: Any):
|
|
"""Sets the last captured SQL error message."""
|
|
_sql_error_state.last_error = str(error)
|
|
|
|
|
|
# 2. Initial Engine Setup
|
|
db_uri = settings.SQLALCHEMY_DB_URI
|
|
|
|
engine = create_engine(
|
|
url = db_uri,
|
|
echo = False,
|
|
pool_use_lifo = True,
|
|
pool_pre_ping = True,
|
|
pool_recycle = settings.DB['pool_recycle'],
|
|
isolation_level = 'READ COMMITTED',
|
|
connect_args = {'connect_timeout': settings.DB['connect_timeout']}
|
|
)
|
|
|
|
log.info('DB SQL Core: Initializing connection...')
|
|
db = None
|
|
try:
|
|
db = engine.connect()
|
|
log.info(f'DB SQL Core: Connected to database: {db_uri}')
|
|
except Exception:
|
|
log.exception('DB SQL Core: Could not connect to database.')
|
|
|
|
|
|
# 3. Connection Management Logic
|
|
def reconnect_db() -> bool:
|
|
"""
|
|
Re-initializes the global database engine and connection using current settings.
|
|
Useful after bootstrapping new credentials from the 'cfg' table.
|
|
"""
|
|
global engine, db, db_uri
|
|
|
|
log.info("DB SQL Core: Refreshing database connection engine...")
|
|
try:
|
|
if engine:
|
|
engine.dispose()
|
|
log.info("DB SQL Core: Disposed of previous database engine.")
|
|
|
|
db_uri = settings.SQLALCHEMY_DB_URI
|
|
engine = create_engine(
|
|
url = db_uri,
|
|
echo = False,
|
|
pool_use_lifo = True,
|
|
pool_pre_ping = True,
|
|
pool_recycle = settings.DB['pool_recycle'],
|
|
isolation_level = 'READ COMMITTED',
|
|
connect_args = {'connect_timeout': settings.DB['connect_timeout']}
|
|
)
|
|
db = engine.connect()
|
|
log.info(f"DB SQL Core: Database connection re-established successfully: {db_uri}")
|
|
return True
|
|
except Exception:
|
|
log.exception("DB SQL Core: FAILED to refresh database connection!")
|
|
return False
|
|
|
|
def sql_connect(current_db, log_lvl: int = logging.INFO) -> bool:
|
|
"""Old compatibility wrapper for disposing the engine."""
|
|
if current_db:
|
|
current_db.engine.dispose()
|
|
log.info('DB SQL Core: Disposed of the current engine via sql_connect.')
|
|
return True
|
|
return False |