database.py•2.67 kB
"""
Database initialization and connection management for the Sectional MCP Panel.
"""
import os
import aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncEngine
from contextlib import asynccontextmanager
from .models import Base
# Database URL - using SQLite for initial implementation
DATABASE_URL = "sqlite+aiosqlite:///./mcp_panel.db"
# Create async engine
engine: AsyncEngine = create_async_engine(
DATABASE_URL,
echo=False,
connect_args={"check_same_thread": False}
)
# Create async session factory
async_session_factory = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)
@asynccontextmanager
async def get_db():
"""Async context manager for database sessions."""
async with async_session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def init_db():
"""Initialize the database, creating tables if they don't exist."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Create initial panel configuration if it doesn't exist
async with get_db() as db:
from sqlalchemy import select
from .models import Panel
result = await db.execute(select(Panel))
panel = result.scalars().first()
if not panel:
panel = Panel(
name="Sectional MCP Panel",
version="0.1.0",
config_schema_version="1.0",
global_settings={
"environmentVars": {
"LOG_LEVEL": "INFO",
"PYTHONUNBUFFERED": "1"
},
"resourceLimits": {
"cpuRequest": None,
"cpuLimit": None,
"memoryRequestMB": None,
"memoryLimitMB": None
},
"runtimeOptions": {
"restartPolicy": "on_failure",
"maxRestarts": 3,
"terminationGracePeriodSeconds": 30
},
"securityContext": {
"runAsUser": None,
"runAsGroup": None,
"readOnlyRootFilesystem": False
}
}
)
db.add(panel)
await db.commit()