"""
Plugin Configuration
Manages Commitizen plugin configuration, discovery, and validation.
"""
from typing import Dict, Any, List, Optional, Type
from dataclasses import dataclass, field
from pathlib import Path
from ..errors import (
handle_errors,
ConfigurationError,
PluginError,
create_success_response,
)
@dataclass
class PluginConfig:
"""Plugin configuration and management."""
plugin_name: str
plugin_class: Optional[Type] = field(default=None)
config: Dict[str, Any] = field(default_factory=dict)
capabilities: Dict[str, bool] = field(default_factory=dict)
adapter_class: Optional[Type] = field(default=None)
def __post_init__(self):
"""Initialize plugin configuration."""
self._discover_plugin()
self._analyze_capabilities()
self._load_adapter()
@handle_errors(log_errors=True)
def _discover_plugin(self):
"""Discover and load plugin class."""
try:
from commitizen import factory
from commitizen.config import BaseConfig
# Create temporary config to load plugin
temp_config = BaseConfig()
temp_config.settings["name"] = self.plugin_name
# Load plugin using commitizen factory
self.plugin_class = factory.committer_factory(temp_config).__class__
except Exception as e:
# Plugin not found or invalid
self.plugin_class = None
raise PluginError(
f"Failed to discover plugin: {self.plugin_name}",
plugin_name=self.plugin_name,
cause=e,
)
@handle_errors(log_errors=True)
def _analyze_capabilities(self):
"""Analyze plugin capabilities."""
if not self.plugin_class:
return
self.capabilities = {
"has_questions": hasattr(self.plugin_class, "questions"),
"has_message": hasattr(self.plugin_class, "message"),
"has_pattern": hasattr(self.plugin_class, "pattern"),
"has_example": hasattr(self.plugin_class, "example"),
"has_schema": hasattr(self.plugin_class, "schema")
or hasattr(self.plugin_class, "schema_pattern"),
"has_bump_pattern": hasattr(self.plugin_class, "bump_pattern"),
"has_bump_map": hasattr(self.plugin_class, "bump_map"),
}
@handle_errors(log_errors=True)
def _load_adapter(self):
"""Load appropriate plugin adapter."""
try:
from ..plugin_adapters import PluginAdapterFactory
self.adapter_class = PluginAdapterFactory.create_adapter(
self.plugin_name
).__class__
except Exception as e:
self.adapter_class = None
raise PluginError(
f"Failed to load adapter for plugin: {self.plugin_name}",
plugin_name=self.plugin_name,
cause=e,
)
@handle_errors(log_errors=True)
def is_valid(self) -> bool:
"""Check if plugin is valid and usable."""
return (
self.plugin_class is not None
and self.capabilities.get("has_message", False)
and self.adapter_class is not None
)
@handle_errors(log_errors=True)
def get_validation_pattern(self) -> Optional[str]:
"""Get validation pattern from plugin."""
if not self.plugin_class or not hasattr(self.plugin_class, "pattern"):
return None
try:
pattern = self.plugin_class.pattern
return pattern if isinstance(pattern, str) else str(pattern)
except Exception as e:
raise PluginError(
f"Failed to get validation pattern for plugin: {self.plugin_name}",
plugin_name=self.plugin_name,
cause=e,
)
@handle_errors(log_errors=True)
def get_supported_types(self) -> List[str]:
"""Get supported commit types from plugin."""
if not self.plugin_class or not hasattr(self.plugin_class, "questions"):
return []
try:
questions = self.plugin_class.questions
if callable(questions):
questions = questions()
for question in questions:
if question.get("name") == "prefix" and "choices" in question:
choices = question["choices"]
return [
choice.get("value", choice)
if isinstance(choice, dict)
else str(choice)
for choice in choices
]
except Exception as e:
raise PluginError(
f"Failed to get supported types for plugin: {self.plugin_name}",
plugin_name=self.plugin_name,
cause=e,
)
return []
@handle_errors(log_errors=True)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary representation."""
return create_success_response(
{
"plugin_name": self.plugin_name,
"plugin_class_name": self.plugin_class.__name__
if self.plugin_class
else None,
"config": self.config,
"capabilities": self.capabilities,
"adapter_class_name": self.adapter_class.__name__
if self.adapter_class
else None,
"is_valid": self.is_valid(),
"validation_pattern": self.get_validation_pattern(),
"supported_types": self.get_supported_types(),
}
)
@handle_errors(log_errors=True)
def discover_available_plugins() -> List[str]:
"""Discover all available Commitizen plugins."""
try:
from commitizen import factory
# This is a simplified discovery - in practice, you might want to
# scan installed packages or use commitizen's plugin discovery
known_plugins = [
"cz_conventional_commits",
"cz_jira",
"cz_customize",
]
return known_plugins
except Exception as e:
raise ConfigurationError("Failed to discover available plugins", cause=e)
@handle_errors(log_errors=True)
def load_plugin_config(plugin_name: str, config: Dict[str, Any] = None) -> PluginConfig:
"""Load configuration for a specific plugin."""
if not plugin_name:
raise ConfigurationError(
"Plugin name cannot be empty", config_key="plugin_name"
)
plugin_config = PluginConfig(plugin_name=plugin_name)
if config:
plugin_config.config = config
return plugin_config