import os from pathlib import Path from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool # Use current working directory if /app doesn't exist base_path = Path("/app") if Path("/app").exists() else Path.cwd() DB_DIR = base_path / "storage" / "db" DB_DIR.mkdir(parents=True, exist_ok=True) SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite" # Optimized engine configuration for better performance engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={ "check_same_thread": False, "timeout": 30, # 30 second timeout "isolation_level": None, # autocommit mode for better performance }, poolclass=StaticPool, pool_pre_ping=True, pool_recycle=3600, # Recycle connections every hour echo=False, # Disable SQL logging in production for performance ) # Enable SQLite optimizations @event.listens_for(engine, "connect") def set_sqlite_pragma(dbapi_connection, connection_record): """Optimize SQLite performance with pragma statements""" cursor = dbapi_connection.cursor() # Enable WAL mode for better concurrency cursor.execute("PRAGMA journal_mode=WAL") # Increase cache size (negative value means KB, positive means pages) cursor.execute("PRAGMA cache_size=-64000") # 64MB cache # Enable foreign keys cursor.execute("PRAGMA foreign_keys=ON") # Optimize synchronous mode cursor.execute("PRAGMA synchronous=NORMAL") # Optimize temp store cursor.execute("PRAGMA temp_store=MEMORY") # Optimize mmap size (256MB) cursor.execute("PRAGMA mmap_size=268435456") cursor.close() SessionLocal = sessionmaker( autocommit=False, autoflush=False, bind=engine, expire_on_commit=False # Prevent lazy loading issues ) def get_db(): db = SessionLocal() try: yield db finally: db.close()