from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from contextlib import contextmanager
class DatabaseAdapter(ABC):
"""
Abstract Base Class that all database adapters must implement.
This ensures a consistent interface for the MCP server regardless of the underlying DB.
"""
@abstractmethod
def __init__(self, config: Dict[str, Any]):
"""Initialize with specific configuration."""
pass
@abstractmethod
@contextmanager
def get_connection(self):
"""Yield a raw connection object intended to be used in a with statement."""
pass
@abstractmethod
def execute_query(self, query: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""
Execute a generic read/write query.
Should return a standard result format:
{
"columns": [...],
"rows": [...],
"row_count": int,
"affected_rows": int
}
"""
pass
@abstractmethod
def list_tables(self) -> List[str]:
"""Return a list of available tables."""
pass
@abstractmethod
def describe_table(self, table_name: str) -> Dict[str, Any]:
"""Return schema information for a specific table."""
pass
@abstractmethod
def get_version(self) -> str:
"""Return the version of the database server."""
pass
