82 lines
2.7 KiB
Python
82 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')
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
|
|
# 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
|
|
|
|
def create_ae_engine(uri: str):
|
|
return create_engine(
|
|
url = uri,
|
|
echo = False,
|
|
pool_size = settings.DB.get('pool_size', 10),
|
|
max_overflow = settings.DB.get('max_overflow', 20),
|
|
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']}
|
|
)
|
|
|
|
engine = create_ae_engine(db_uri)
|
|
|
|
# DEPRECATED: Global shared 'db' connection. Use engine.connect() in context managers instead.
|
|
# Keeping for legacy compatibility but will phase out usage in crud lib.
|
|
db = engine.connect()
|
|
|
|
log.info('DB SQL Core: Initializing engine...')
|
|
|
|
|
|
# 3. Connection Management Logic
|
|
def reconnect_db() -> bool:
|
|
"""
|
|
Re-initializes the global database engine 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_ae_engine(db_uri)
|
|
db = engine.connect()
|
|
|
|
safe_uri = db_uri.split('@')[-1] if '@' in db_uri else db_uri
|
|
log.info(f"DB SQL Core: Database engine re-established successfully: {safe_uri}")
|
|
return True
|
|
except Exception:
|
|
log.exception("DB SQL Core: FAILED to refresh database engine!")
|
|
return False
|
|
|
|
def sql_connect(current_db=None, log_lvl: int = logging.INFO) -> bool:
|
|
"""Refreshes the global database connection."""
|
|
log.setLevel(log_lvl)
|
|
log.info('DB SQL Core: Refreshing database connection via sql_connect...')
|
|
return reconnect_db() |