""" Separate database for seed data storage This keeps seed SQL content isolated from the main Calendar database """ import os import logging from datetime import datetime from sqlalchemy import Column, DateTime, Integer, String, Text, create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker logger = logging.getLogger(__name__) # Separate Base for seed database SeedBase = declarative_base() # Seed database path - stored in mcp_databases folder for persistence SEED_DATABASE_PATH = os.getenv("SEED_DATABASE_PATH", "./mcp_databases/seed_store.db") SEED_DATABASE_URL = f"sqlite:///{SEED_DATABASE_PATH}" # Create engine for seed database seed_engine = create_engine( SEED_DATABASE_URL, connect_args={"check_same_thread": False}, echo=False ) # Create session factory for seed database SeedSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=seed_engine) class SeedData(SeedBase): """Model to store seed SQL content for databases""" __tablename__ = "seed_data" id = Column(Integer, primary_key=True, index=True) database_id = Column(String(255), unique=True, nullable=False, index=True) name = Column(String(255), nullable=True) description = Column(Text, nullable=True) sql_content = Column(Text, nullable=False) created_at = Column(DateTime, default=datetime.utcnow, nullable=False) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False) def __repr__(self): return f"" def init_seed_database(): """Initialize the seed database and create tables""" # Ensure the mcp_databases directory exists db_dir = os.path.dirname(SEED_DATABASE_PATH) if db_dir and not os.path.exists(db_dir): os.makedirs(db_dir, exist_ok=True) logger.info(f"Created directory: {db_dir}") logger.info(f"Initializing seed database at {SEED_DATABASE_PATH}") SeedBase.metadata.create_all(bind=seed_engine) logger.info("Seed database initialized successfully") def get_seed_session(): """Get a session for the seed database""" return SeedSessionLocal() __all__ = ["SeedData", "get_seed_session", "init_seed_database"]