"""Configuration management for PostgreSQL MCP Server"""
import os
from typing import Optional
from pydantic import BaseSettings, Field, validator
from pydantic_settings import BaseSettings as PydanticBaseSettings
class DatabaseConfig(PydanticBaseSettings):
"""Database connection configuration"""
host: str = Field(default="localhost", description="PostgreSQL host")
port: int = Field(default=5432, description="PostgreSQL port")
database: str = Field(..., description="Database name")
username: str = Field(..., description="Database username")
password: str = Field(..., description="Database password")
ssl_mode: str = Field(default="prefer", description="SSL mode")
# Connection pool settings
min_connections: int = Field(default=1, description="Minimum pool connections")
max_connections: int = Field(default=10, description="Maximum pool connections")
@validator('port')
def validate_port(cls, v):
if not 1 <= v <= 65535:
raise ValueError('Port must be between 1 and 65535')
return v
@property
def connection_string(self) -> str:
"""Generate PostgreSQL connection string"""
return (
f"postgresql://{self.username}:{self.password}"
f"@{self.host}:{self.port}/{self.database}"
f"?sslmode={self.ssl_mode}"
)
class Config:
env_prefix = "POSTGRES_"
env_file = ".env"
class ServerConfig(PydanticBaseSettings):
"""MCP Server configuration"""
name: str = Field(default="postgresql-mcp-server", description="Server name")
version: str = Field(default="1.0.0", description="Server version")
# Security settings
max_query_time: int = Field(default=30, description="Maximum query execution time (seconds)")
max_rows: int = Field(default=1000, description="Maximum rows returned per query")
allowed_schemas: Optional[list[str]] = Field(default=None, description="Allowed database schemas")
# Logging
log_level: str = Field(default="INFO", description="Logging level")
log_queries: bool = Field(default=True, description="Log executed queries")
@validator('log_level')
def validate_log_level(cls, v):
valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
if v.upper() not in valid_levels:
raise ValueError(f'Log level must be one of: {valid_levels}')
return v.upper()
class Config:
env_prefix = "MCP_"
env_file = ".env"
def load_config() -> tuple[DatabaseConfig, ServerConfig]:
"""Load configuration from environment variables and .env file"""
db_config = DatabaseConfig()
server_config = ServerConfig()
return db_config, server_config