"""Configuration management for DP-MCP Server."""
import os
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
class PostgreSQLConfig(BaseModel):
"""PostgreSQL configuration."""
host: str = Field(default="localhost")
port: int = Field(default=5432)
user: Optional[str] = Field(default=None)
password: Optional[str] = Field(default=None)
database: Optional[str] = Field(default=None)
ssl_mode: str = Field(default="prefer")
class MinIOConfig(BaseModel):
"""MinIO configuration."""
endpoint: str = Field(default="localhost:9000")
access_key: Optional[str] = Field(default=None)
secret_key: Optional[str] = Field(default=None)
secure: bool = Field(default=False)
default_bucket: str = Field(default="default-bucket")
class ServerConfig(BaseModel):
"""MCP Server configuration."""
host: str = Field(default="127.0.0.1")
port: int = Field(default=8888)
debug: bool = Field(default=False)
class Config(BaseModel):
"""Complete configuration."""
postgres: PostgreSQLConfig
minio: MinIOConfig
server: ServerConfig
def get_config() -> Config:
"""Load configuration from environment variables."""
return Config(
postgres=PostgreSQLConfig(
host=os.getenv("POSTGRES_HOST", "localhost"),
port=int(os.getenv("POSTGRES_PORT", 5432)),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD"),
database=os.getenv("POSTGRES_DATABASE"),
ssl_mode=os.getenv("POSTGRES_SSL_MODE", "prefer")
),
minio=MinIOConfig(
endpoint=os.getenv("MINIO_ENDPOINT", "localhost:9000"),
access_key=os.getenv("MINIO_ACCESS_KEY"),
secret_key=os.getenv("MINIO_SECRET_KEY"),
secure=os.getenv("MINIO_SECURE", "false").lower() == "true",
default_bucket=os.getenv("MINIO_DEFAULT_BUCKET", "default-bucket")
),
server=ServerConfig(
host=os.getenv("MCP_SERVER_HOST", "127.0.0.1"),
port=int(os.getenv("MCP_SERVER_PORT", 8888)),
debug=os.getenv("DEBUG", "false").lower() == "true"
)
)
def validate_config(config: Config) -> Dict[str, Any]:
"""Validate configuration and return validation results."""
errors = []
# Validate PostgreSQL config
if not config.postgres.user:
errors.append("POSTGRES_USER is required")
if not config.postgres.password:
errors.append("POSTGRES_PASSWORD is required")
if not config.postgres.database:
errors.append("POSTGRES_DATABASE is required")
# Validate MinIO config
if not config.minio.access_key:
errors.append("MINIO_ACCESS_KEY is required")
if not config.minio.secret_key:
errors.append("MINIO_SECRET_KEY is required")
return {
"valid": len(errors) == 0,
"errors": errors
}
if __name__ == "__main__":
config = get_config()
validation = validate_config(config)
print("Configuration Status:")
if validation["valid"]:
print("✅ Configuration is valid!")
else:
print("❌ Configuration errors:")
for error in validation["errors"]:
print(f" • {error}")
