Coverage for fastblocks / mcp / configuration.py: 46%
319 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-26 03:30 -0800
1"""Advanced adapter configuration management for FastBlocks MCP system."""
3import json
4import os
5from contextlib import suppress
6from dataclasses import dataclass, field
7from datetime import datetime
8from enum import Enum
9from pathlib import Path
10from typing import Any
11from uuid import uuid4
13import yaml
14from pydantic import BaseModel, Field, ValidationError, field_validator
16from .discovery import AdapterInfo
17from .registry import AdapterRegistry
20class ConfigurationProfile(str, Enum):
21 """Configuration deployment profiles."""
23 DEVELOPMENT = "development"
24 STAGING = "staging"
25 PRODUCTION = "production"
28class ConfigurationStatus(str, Enum):
29 """Configuration validation status."""
31 VALID = "valid"
32 WARNING = "warning"
33 ERROR = "error"
34 UNKNOWN = "unknown"
37@dataclass
38class EnvironmentVariable:
39 """Environment variable configuration."""
41 name: str
42 value: str | None = None
43 required: bool = True
44 description: str = ""
45 secret: bool = False
46 default: str | None = None
47 validator_pattern: str | None = None
50@dataclass
51class AdapterConfiguration:
52 """Individual adapter configuration."""
54 name: str
55 enabled: bool = True
56 settings: dict[str, Any] = field(default_factory=dict)
57 environment_variables: list[EnvironmentVariable] = field(default_factory=list)
58 dependencies: set[str] = field(default_factory=set)
59 profile_overrides: dict[ConfigurationProfile, dict[str, Any]] = field(
60 default_factory=dict
61 )
62 health_check_config: dict[str, Any] = field(default_factory=dict)
63 metadata: dict[str, Any] = field(default_factory=dict)
66class ConfigurationSchema(BaseModel):
67 """Pydantic schema for configuration validation."""
69 version: str = "1.0"
70 profile: ConfigurationProfile = ConfigurationProfile.DEVELOPMENT
71 created_at: datetime = Field(default_factory=datetime.now)
72 updated_at: datetime = Field(default_factory=datetime.now)
73 adapters: dict[str, AdapterConfiguration] = Field(default_factory=dict)
74 global_settings: dict[str, Any] = Field(default_factory=dict)
75 global_environment: list[EnvironmentVariable] = Field(default_factory=list)
77 @field_validator("adapters", mode="before")
78 @classmethod
79 def validate_adapters(cls, v: Any) -> dict[str, AdapterConfiguration]:
80 if isinstance(v, dict):
81 # Convert dict values to AdapterConfiguration objects if needed
82 result: dict[str, AdapterConfiguration] = {}
83 for key, value in v.items():
84 if isinstance(value, dict):
85 result[key] = AdapterConfiguration(name=key, **value)
86 else:
87 result[key] = value
88 return result
89 # v should already be dict[str, AdapterConfiguration] if not dict
90 return v if isinstance(v, dict) else {}
92 class Config:
93 arbitrary_types_allowed = True
96@dataclass
97class ConfigurationValidationResult:
98 """Result of configuration validation."""
100 status: ConfigurationStatus
101 errors: list[str] = field(default_factory=list)
102 warnings: list[str] = field(default_factory=list)
103 info: dict[str, Any] = field(default_factory=dict)
104 adapter_results: dict[str, dict[str, Any]] = field(default_factory=dict)
107@dataclass
108class ConfigurationBackup:
109 """Configuration backup metadata."""
111 id: str
112 name: str
113 description: str
114 created_at: datetime
115 profile: ConfigurationProfile
116 file_path: Path
117 checksum: str
120class ConfigurationManager:
121 """Advanced configuration management for FastBlocks adapters."""
123 def __init__(self, registry: AdapterRegistry, base_path: Path | None = None):
124 """Initialize configuration manager."""
125 self.registry = registry
126 self.base_path = base_path or Path.cwd() / ".fastblocks"
127 self.config_dir = self.base_path / "config"
128 self.backup_dir = self.base_path / "backups"
129 self.templates_dir = self.base_path / "templates"
131 # Ensure directories exist
132 for directory in (self.config_dir, self.backup_dir, self.templates_dir):
133 directory.mkdir(parents=True, exist_ok=True)
135 async def initialize(self) -> None:
136 """Initialize configuration manager."""
137 await self.registry.initialize()
138 await self._ensure_default_templates()
140 async def get_available_adapters(self) -> dict[str, AdapterInfo]:
141 """Get all available adapters for configuration."""
142 return await self.registry.list_available_adapters()
144 async def get_adapter_configuration_schema(
145 self, adapter_name: str
146 ) -> dict[str, Any]:
147 """Get configuration schema for a specific adapter."""
148 adapter_info = await self.registry.get_adapter_info(adapter_name)
149 if not adapter_info:
150 raise ValueError(f"Adapter '{adapter_name}' not found")
152 # Build base schema
153 schema = self._build_base_schema(adapter_name, adapter_info)
155 # Try to introspect adapter settings
156 with suppress(Exception):
157 adapter = await self.registry.get_adapter(adapter_name)
158 self._introspect_adapter_settings(adapter, schema)
160 return schema
162 def _build_base_schema(
163 self, adapter_name: str, adapter_info: AdapterInfo
164 ) -> dict[str, Any]:
165 """Build base schema structure."""
166 return {
167 "name": adapter_name,
168 "description": adapter_info.description,
169 "category": adapter_info.category,
170 "required_settings": [],
171 "optional_settings": [],
172 "environment_variables": [],
173 "dependencies": [],
174 }
176 def _introspect_adapter_settings(
177 self, adapter: Any, schema: dict[str, Any]
178 ) -> None:
179 """Introspect adapter settings and populate schema."""
180 if not adapter or not hasattr(adapter, "settings"):
181 return
183 settings = adapter.settings
184 if not hasattr(settings, "__dict__"):
185 return
187 # Categorize settings by requirement
188 categorized = self._categorize_settings(settings.__dict__)
189 schema["required_settings"] = categorized["required"]
190 schema["optional_settings"] = categorized["optional"]
192 def _categorize_settings(
193 self, settings_dict: dict[str, Any]
194 ) -> dict[str, list[dict[str, Any]]]:
195 """Categorize settings into required and optional."""
196 categorized: dict[str, list[dict[str, Any]]] = {
197 "required": [],
198 "optional": [],
199 }
201 for key, value in settings_dict.items():
202 if key.startswith("_"):
203 continue
205 setting_info = {
206 "name": key,
207 "type": type(value).__name__,
208 "default": value,
209 "required": value is None,
210 }
212 category = "required" if setting_info["required"] else "optional"
213 categorized[category].append(setting_info)
215 return categorized
217 async def create_configuration(
218 self,
219 profile: ConfigurationProfile = ConfigurationProfile.DEVELOPMENT,
220 adapters: list[str] | None = None,
221 ) -> ConfigurationSchema:
222 """Create a new configuration."""
223 config = ConfigurationSchema(profile=profile)
225 if adapters:
226 for adapter_name in adapters:
227 adapter_config = await self._create_adapter_configuration(adapter_name)
228 config.adapters[adapter_name] = adapter_config
230 return config
232 async def _create_adapter_configuration(
233 self, adapter_name: str
234 ) -> AdapterConfiguration:
235 """Create configuration for a specific adapter."""
236 schema = await self.get_adapter_configuration_schema(adapter_name)
238 adapter_config = AdapterConfiguration(name=adapter_name)
240 # Set up environment variables based on schema
241 for setting in schema.get("required_settings", []):
242 env_var = EnvironmentVariable(
243 name=f"FB_{adapter_name.upper()}_{setting['name'].upper()}",
244 required=True,
245 description=f"Required setting for {adapter_name}: {setting['name']}",
246 )
247 adapter_config.environment_variables.append(env_var)
249 for setting in schema.get("optional_settings", []):
250 env_var = EnvironmentVariable(
251 name=f"FB_{adapter_name.upper()}_{setting['name'].upper()}",
252 required=False,
253 default=str(setting.get("default", "")),
254 description=f"Optional setting for {adapter_name}: {setting['name']}",
255 )
256 adapter_config.environment_variables.append(env_var)
258 return adapter_config
260 async def validate_configuration(
261 self, config: ConfigurationSchema
262 ) -> ConfigurationValidationResult:
263 """Validate a configuration comprehensively."""
264 result = ConfigurationValidationResult(status=ConfigurationStatus.VALID)
266 try:
267 # Validate configuration schema
268 config_dict = (
269 config.model_dump()
270 if hasattr(config, "model_dump")
271 else config.__dict__
272 )
273 ConfigurationSchema(**config_dict)
274 except ValidationError as e:
275 result.status = ConfigurationStatus.ERROR
276 result.errors.extend([str(error) for error in e.errors()])
277 except Exception as e:
278 result.status = ConfigurationStatus.ERROR
279 result.errors.append(f"Configuration validation error: {e}")
281 # Validate individual adapters
282 for adapter_name, adapter_config in config.adapters.items():
283 adapter_result = await self._validate_adapter_configuration(
284 adapter_name, adapter_config
285 )
286 result.adapter_results[adapter_name] = adapter_result
288 if adapter_result.get("errors"):
289 result.errors.extend(
290 f"{adapter_name}: {error}" for error in adapter_result["errors"]
291 )
292 result.status = ConfigurationStatus.ERROR
294 if adapter_result.get("warnings"):
295 result.warnings.extend(
296 f"{adapter_name}: {warning}"
297 for warning in adapter_result["warnings"]
298 )
299 if result.status == ConfigurationStatus.VALID:
300 result.status = ConfigurationStatus.WARNING
302 # Validate dependencies
303 await self._validate_dependencies(config, result)
305 # Validate environment variables
306 await self._validate_environment_variables(config, result)
308 return result
310 async def _validate_adapter_configuration(
311 self, adapter_name: str, adapter_config: AdapterConfiguration
312 ) -> dict[str, Any]:
313 """Validate individual adapter configuration."""
314 validation_result = await self.registry.validate_adapter(adapter_name)
316 result = {
317 "valid": validation_result.get("valid", False),
318 "errors": validation_result.get("errors", []),
319 "warnings": validation_result.get("warnings", []),
320 "info": validation_result.get("info", {}),
321 }
323 # Additional configuration-specific validations
324 if adapter_config.enabled:
325 # Check required environment variables
326 for env_var in adapter_config.environment_variables:
327 if env_var.required and not env_var.value and not env_var.default:
328 if env_var.name not in os.environ:
329 result["warnings"].append(
330 f"Required environment variable {env_var.name} is not set"
331 )
333 return result
335 async def _validate_dependencies(
336 self, config: ConfigurationSchema, result: ConfigurationValidationResult
337 ) -> None:
338 """Validate adapter dependencies."""
339 enabled_adapters = {
340 name for name, adapter in config.adapters.items() if adapter.enabled
341 }
343 for adapter_name, adapter_config in config.adapters.items():
344 if not adapter_config.enabled:
345 continue
347 for dependency in adapter_config.dependencies:
348 if dependency not in enabled_adapters:
349 result.errors.append(
350 f"Adapter '{adapter_name}' depends on '{dependency}' which is not enabled"
351 )
352 result.status = ConfigurationStatus.ERROR
354 def _check_duplicate_env_vars(
355 self,
356 config: ConfigurationSchema,
357 result: ConfigurationValidationResult,
358 all_env_vars: set[str],
359 ) -> None:
360 """Check for duplicate environment variable names."""
361 for adapter_config in config.adapters.values():
362 for env_var in adapter_config.environment_variables:
363 if env_var.name in all_env_vars:
364 result.warnings.append(
365 f"Duplicate environment variable: {env_var.name}"
366 )
367 all_env_vars.add(env_var.name)
369 def _is_env_var_missing(self, env_var: EnvironmentVariable) -> bool:
370 """Check if a required environment variable is missing."""
371 return (
372 env_var.required
373 and not env_var.value
374 and not env_var.default
375 and env_var.name not in os.environ
376 )
378 def _check_missing_required_vars(
379 self, config: ConfigurationSchema, result: ConfigurationValidationResult
380 ) -> None:
381 """Check for missing required environment variables."""
382 for adapter_name, adapter_config in config.adapters.items():
383 if not adapter_config.enabled:
384 continue
386 for env_var in adapter_config.environment_variables:
387 if self._is_env_var_missing(env_var):
388 result.warnings.append(
389 f"Required environment variable {env_var.name} for {adapter_name} is not set"
390 )
392 async def _validate_environment_variables(
393 self, config: ConfigurationSchema, result: ConfigurationValidationResult
394 ) -> None:
395 """Validate environment variable configuration."""
396 all_env_vars: set[str] = set()
398 # Check for duplicate environment variable names
399 self._check_duplicate_env_vars(config, result, all_env_vars)
401 # Check for missing required variables
402 self._check_missing_required_vars(config, result)
404 async def save_configuration(
405 self, config: ConfigurationSchema, name: str | None = None
406 ) -> Path:
407 """Save configuration to YAML file."""
408 if not name:
409 name = f"{config.profile.value}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
411 config_file = self.config_dir / f"{name}.yaml"
413 # Convert to serializable dict
414 config_dict = self._serialize_configuration(config)
416 with config_file.open("w") as f:
417 yaml.dump(config_dict, f, default_flow_style=False, sort_keys=False)
419 return config_file
421 async def load_configuration(self, name_or_path: str | Path) -> ConfigurationSchema:
422 """Load configuration from YAML file."""
423 if isinstance(name_or_path, str):
424 config_file = self.config_dir / f"{name_or_path}.yaml"
425 if not config_file.exists():
426 config_file = Path(name_or_path)
427 else:
428 config_file = name_or_path
430 if not config_file.exists():
431 raise FileNotFoundError(f"Configuration file not found: {config_file}")
433 with config_file.open() as f:
434 config_dict = yaml.safe_load(f)
436 return self._deserialize_configuration(config_dict)
438 def _serialize_configuration(self, config: ConfigurationSchema) -> dict[str, Any]:
439 """Convert configuration to serializable dictionary."""
440 result = {
441 "version": config.version,
442 "profile": config.profile.value,
443 "created_at": config.created_at.isoformat(),
444 "updated_at": config.updated_at.isoformat(),
445 "global_settings": config.global_settings,
446 "global_environment": [
447 {
448 "name": env_var.name,
449 "value": env_var.value,
450 "required": env_var.required,
451 "description": env_var.description,
452 "secret": env_var.secret,
453 "default": env_var.default,
454 "validator_pattern": env_var.validator_pattern,
455 }
456 for env_var in config.global_environment
457 ],
458 "adapters": {},
459 }
461 adapters_dict: dict[str, Any] = {}
462 for adapter_name, adapter_config in config.adapters.items():
463 adapters_dict[adapter_name] = {
464 "enabled": adapter_config.enabled,
465 "settings": adapter_config.settings,
466 "environment_variables": [
467 {
468 "name": env_var.name,
469 "value": env_var.value,
470 "required": env_var.required,
471 "description": env_var.description,
472 "secret": env_var.secret,
473 "default": env_var.default,
474 "validator_pattern": env_var.validator_pattern,
475 }
476 for env_var in adapter_config.environment_variables
477 ],
478 "dependencies": list(adapter_config.dependencies),
479 "profile_overrides": {
480 profile.value: overrides
481 for profile, overrides in adapter_config.profile_overrides.items()
482 },
483 "health_check_config": adapter_config.health_check_config,
484 "metadata": adapter_config.metadata,
485 }
487 result["adapters"] = adapters_dict
488 return result
490 def _deserialize_configuration(
491 self, config_dict: dict[str, Any]
492 ) -> ConfigurationSchema:
493 """Convert dictionary to configuration object."""
494 # Convert string dates back to datetime objects
495 if "created_at" in config_dict:
496 config_dict["created_at"] = datetime.fromisoformat(
497 config_dict["created_at"]
498 )
499 if "updated_at" in config_dict:
500 config_dict["updated_at"] = datetime.fromisoformat(
501 config_dict["updated_at"]
502 )
504 # Convert profile string to enum
505 if "profile" in config_dict:
506 config_dict["profile"] = ConfigurationProfile(config_dict["profile"])
508 # Convert global environment variables
509 global_env = [
510 EnvironmentVariable(**env_data)
511 for env_data in config_dict.get("global_environment", [])
512 ]
513 config_dict["global_environment"] = global_env
515 # Convert adapter configurations
516 adapters = {}
517 for adapter_name, adapter_data in config_dict.get("adapters", {}).items():
518 # Convert environment variables
519 env_vars = [
520 EnvironmentVariable(**env_data)
521 for env_data in adapter_data.get("environment_variables", [])
522 ]
523 adapter_data["environment_variables"] = env_vars
525 # Convert profile overrides
526 profile_overrides = {}
527 for profile_str, overrides in adapter_data.get(
528 "profile_overrides", {}
529 ).items():
530 profile_overrides[ConfigurationProfile(profile_str)] = overrides
531 adapter_data["profile_overrides"] = profile_overrides
533 # Convert dependencies to set
534 adapter_data["dependencies"] = set(adapter_data.get("dependencies", []))
536 adapters[adapter_name] = AdapterConfiguration(
537 name=adapter_name, **adapter_data
538 )
540 config_dict["adapters"] = adapters
542 return ConfigurationSchema(**config_dict)
544 async def generate_environment_file(
545 self, config: ConfigurationSchema, output_path: Path | None = None
546 ) -> Path:
547 """Generate .env file from configuration."""
548 if not output_path:
549 output_path = self.base_path / f".env.{config.profile.value}"
551 # Add global environment variables
552 env_vars = [
553 self._format_env_var(env_var) for env_var in config.global_environment
554 ]
556 # Add adapter environment variables
557 for adapter_name, adapter_config in config.adapters.items():
558 if not adapter_config.enabled:
559 continue
561 env_vars.append(f"\n# {adapter_name.upper()} ADAPTER")
562 for env_var in adapter_config.environment_variables:
563 env_vars.append(self._format_env_var(env_var))
565 with output_path.open("w") as f:
566 f.write("\n".join(env_vars))
568 return output_path
570 def _format_env_var(self, env_var: EnvironmentVariable) -> str:
571 """Format environment variable for .env file."""
572 lines = []
574 if env_var.description:
575 lines.append(f"# {env_var.description}")
577 if env_var.required:
578 lines.append("# REQUIRED")
580 value = env_var.value or env_var.default or ""
581 if env_var.secret and value:
582 value = "***REDACTED***"
584 lines.append(f"{env_var.name}={value}")
586 return "\n".join(lines)
588 async def backup_configuration(
589 self, config: ConfigurationSchema, name: str, description: str = ""
590 ) -> ConfigurationBackup:
591 """Create a backup of the configuration."""
592 backup_id = str(uuid4())
593 backup_file = self.backup_dir / f"{backup_id}_{name}.yaml"
595 # Save configuration
596 config_dict = self._serialize_configuration(config)
597 with backup_file.open("w") as f:
598 yaml.dump(config_dict, f, default_flow_style=False)
600 # Calculate checksum
601 import hashlib
603 with backup_file.open("rb") as fb:
604 checksum = hashlib.sha256(fb.read()).hexdigest()
606 backup = ConfigurationBackup(
607 id=backup_id,
608 name=name,
609 description=description,
610 created_at=datetime.now(),
611 profile=config.profile,
612 file_path=backup_file,
613 checksum=checksum,
614 )
616 # Save backup metadata
617 metadata_file = self.backup_dir / f"{backup_id}_metadata.json"
618 with metadata_file.open("w") as f:
619 json.dump(
620 {
621 "id": backup.id,
622 "name": backup.name,
623 "description": backup.description,
624 "created_at": backup.created_at.isoformat(),
625 "profile": backup.profile.value,
626 "file_path": str(backup.file_path),
627 "checksum": backup.checksum,
628 },
629 f,
630 indent=2,
631 )
633 return backup
635 async def list_backups(self) -> list[ConfigurationBackup]:
636 """List all configuration backups."""
637 backups = []
639 for metadata_file in self.backup_dir.glob("*_metadata.json"):
640 try:
641 with metadata_file.open() as f:
642 data = json.load(f)
644 backup = ConfigurationBackup(
645 id=data["id"],
646 name=data["name"],
647 description=data["description"],
648 created_at=datetime.fromisoformat(data["created_at"]),
649 profile=ConfigurationProfile(data["profile"]),
650 file_path=Path(data["file_path"]),
651 checksum=data["checksum"],
652 )
654 # Verify file still exists
655 if backup.file_path.exists():
656 backups.append(backup)
657 except Exception:
658 # Skip corrupted metadata files
659 continue
661 return sorted(backups, key=lambda b: b.created_at, reverse=True)
663 async def restore_backup(self, backup_id: str) -> ConfigurationSchema:
664 """Restore configuration from backup."""
665 backups = await self.list_backups()
666 backup = next((b for b in backups if b.id == backup_id), None)
668 if not backup:
669 raise ValueError(f"Backup '{backup_id}' not found")
671 return await self.load_configuration(backup.file_path)
673 async def _ensure_default_templates(self) -> None:
674 """Ensure default configuration templates exist."""
675 templates = {
676 "minimal.yaml": self._create_minimal_template(),
677 "development.yaml": self._create_development_template(),
678 "production.yaml": self._create_production_template(),
679 }
681 for template_name, template_config in templates.items():
682 template_file = self.templates_dir / template_name
683 if not template_file.exists():
684 config_dict = self._serialize_configuration(template_config)
685 with template_file.open("w") as f:
686 yaml.dump(config_dict, f, default_flow_style=False)
688 def _create_minimal_template(self) -> ConfigurationSchema:
689 """Create minimal configuration template."""
690 return ConfigurationSchema(
691 profile=ConfigurationProfile.DEVELOPMENT,
692 global_settings={"debug": True, "log_level": "INFO"},
693 )
695 def _create_development_template(self) -> ConfigurationSchema:
696 """Create development configuration template."""
697 config = ConfigurationSchema(
698 profile=ConfigurationProfile.DEVELOPMENT,
699 global_settings={"debug": True, "log_level": "DEBUG", "hot_reload": True},
700 )
702 # Add common development adapters
703 config.adapters["app"] = AdapterConfiguration(
704 name="app", enabled=True, settings={"debug": True}
705 )
707 return config
709 def _create_production_template(self) -> ConfigurationSchema:
710 """Create production configuration template."""
711 config = ConfigurationSchema(
712 profile=ConfigurationProfile.PRODUCTION,
713 global_settings={
714 "debug": False,
715 "log_level": "WARNING",
716 "hot_reload": False,
717 },
718 )
720 return config