Spaces:
Sleeping
Sleeping
File size: 2,411 Bytes
01d5a5d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from contextlib import contextmanager
import logging
from lpm_kernel.configs.config import Config
import os
logger = logging.getLogger(__name__)
class Base(DeclarativeBase):
pass
class DatabaseSession:
_instance = None
_engine = None
_session_factory = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls.initialize()
return cls._instance
@classmethod
def initialize(cls):
"""Initialize database engine and session factory"""
if not cls._engine:
try:
config = Config.from_env()
db_config = config.database.to_dict()
# Ensure database directory exists
db_dir = os.path.dirname(db_config['db_file'])
if not os.path.exists(db_dir):
os.makedirs(db_dir)
# Build SQLite connection URL
db_url = f"sqlite:///{db_config['db_file']}"
cls._engine = create_engine(
db_url,
echo=False,
pool_pre_ping=True,
pool_recycle=db_config['pool_recycle'],
pool_size=db_config['maxsize'],
max_overflow=20,
)
cls._session_factory = sessionmaker(bind=cls._engine)
logger.info("SQLite database engine and session factory initialized")
except Exception as e:
logger.error(f"Failed to initialize database: {str(e)}")
raise
@classmethod
@contextmanager
def session(cls):
"""Get database session"""
if not cls._session_factory:
cls.initialize()
session = cls._session_factory()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
@classmethod
def close(cls):
"""Close database engine - should only be called when shutting down the application"""
if cls._engine:
cls._engine.dispose()
cls._engine = None
cls._session_factory = None
logger.info("Database engine closed")
|