pylord/API-BFSI

1"""2Database Configuration and Connection Management3Uses centralized configuration from config.py4"""5 6from sqlalchemy import create_engine7from sqlalchemy.ext.declarative import declarative_base8from sqlalchemy.orm import sessionmaker9from config import settings10import logging11 12logger = logging.getLogger(__name__)13 14# Database URL from settings15SQLALCHEMY_DATABASE_URL = settings.database_url16 17# Create engine with connection pooling18engine = create_engine(19    SQLALCHEMY_DATABASE_URL,20    pool_size=settings.DB_POOL_SIZE,21    max_overflow=settings.DB_MAX_OVERFLOW,22    pool_timeout=settings.DB_POOL_TIMEOUT,23    pool_recycle=settings.DB_POOL_RECYCLE,24    pool_pre_ping=True,  # Verify connections before using25    echo=settings.DEBUG  # Log SQL queries in debug mode26)27 28# Session factory29SessionLocal = sessionmaker(30    autocommit=False,31    autoflush=False,32    bind=engine33)34 35# Base class for ORM models36Base = declarative_base()37 38 39def get_db():40    """41    Database session dependency for FastAPI42    Yields a database session and ensures it's closed after use43    """44    db = SessionLocal()45    try:46        yield db47    except Exception as e:48        logger.error(f"Database session error: {e}")49        db.rollback()50        raise51    finally:52        db.close()53 54 55def init_db():56    """57    Initialize database tables58    Creates all tables if they don't exist59    """60    try:61        Base.metadata.create_all(bind=engine)62        logger.info("✅ Database tables initialized successfully")63    except Exception as e:64        logger.error(f"❌ Database initialization failed: {e}")65        raise66 67 68def check_db_connection():69    """70    Check if database connection is working71    Returns True if connection successful, False otherwise72    """73    try:74        with engine.connect() as connection:75            connection.execute("SELECT 1")76        logger.info("✅ Database connection successful")77        return True78    except Exception as e:79        logger.error(f"❌ Database connection failed: {e}")80        return False81 82 83def get_db_info():84    """85    Get database information for monitoring86    """87    try:88        with engine.connect() as connection:89            result = connection.execute("SELECT version()")90            version = result.fetchone()[0]91            92            return {93                "status": "connected",94                "database": settings.DB_NAME,95                "host": settings.DB_HOST,96                "port": settings.DB_PORT,97                "version": version,98                "pool_size": settings.DB_POOL_SIZE,99                "max_overflow": settings.DB_MAX_OVERFLOW100            }101    except Exception as e:102        return {103            "status": "error",104            "error": str(e)105        }106 107 108if __name__ == "__main__":109    # Test database connection when run directly110    print("Testing database connection...")111    if check_db_connection():112        print("✅ Database connection successful")113        info = get_db_info()114        print(f"Database: {info.get('database')}")115        print(f"Host: {info.get('host')}:{info.get('port')}")116        print(f"Version: {info.get('version')}")117    else:118        print("❌ Database connection failed")