diff --git a/README.md b/README.md index b2b08d5..f6717f6 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,29 @@ Add to your Claude Desktop configuration file: "command": "uvx", "args": ["--from", "pypi-query-mcp-server", "pypi-query-mcp"], "env": { - "PYPI_INDEX_URL": "https://pypi.org/simple/", - "CACHE_TTL": "3600" + "PYPI_INDEX_URL": "https://pypi.org/pypi", + "PYPI_INDEX_URLS": "https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/", + "PYPI_CACHE_TTL": "3600", + "PYPI_LOG_LEVEL": "INFO" + } + } + } +} +``` + +#### With Private Repository +```json +{ + "mcpServers": { + "pypi-query": { + "command": "uvx", + "args": ["--from", "pypi-query-mcp-server", "pypi-query-mcp"], + "env": { + "PYPI_INDEX_URL": "https://pypi.org/pypi", + "PYPI_PRIVATE_PYPI_URL": "https://private.pypi.company.com", + "PYPI_PRIVATE_PYPI_USERNAME": "your_username", + "PYPI_PRIVATE_PYPI_PASSWORD": "your_password", + "PYPI_CACHE_TTL": "3600" } } } @@ -127,11 +148,38 @@ Add to your Windsurf MCP configuration (`~/.codeium/windsurf/mcp_config.json`): ### Environment Variables -- `PYPI_INDEX_URL`: PyPI index URL (default: https://pypi.org/simple/) -- `CACHE_TTL`: Cache time-to-live in seconds (default: 3600) -- `PRIVATE_PYPI_URL`: Private PyPI repository URL (optional) -- `PRIVATE_PYPI_USERNAME`: Private PyPI username (optional) -- `PRIVATE_PYPI_PASSWORD`: Private PyPI password (optional) +#### Basic Configuration +- `PYPI_INDEX_URL`: Primary PyPI index URL (default: https://pypi.org/pypi) +- `PYPI_CACHE_TTL`: Cache time-to-live in seconds (default: 3600) +- `PYPI_LOG_LEVEL`: Logging level (default: INFO) +- `PYPI_REQUEST_TIMEOUT`: HTTP request timeout in seconds (default: 30.0) + +#### Multiple Mirror Sources Support +- `PYPI_INDEX_URLS`: Additional PyPI index URLs (comma-separated, optional) +- `PYPI_EXTRA_INDEX_URLS`: Extra PyPI index URLs for fallback (comma-separated, optional) + +#### Private Repository Support +- `PYPI_PRIVATE_PYPI_URL`: Private PyPI repository URL (optional) +- `PYPI_PRIVATE_PYPI_USERNAME`: Private PyPI username (optional) +- `PYPI_PRIVATE_PYPI_PASSWORD`: Private PyPI password (optional) + +#### Advanced Dependency Analysis +- `PYPI_DEPENDENCY_MAX_DEPTH`: Maximum depth for recursive dependency analysis (default: 5) +- `PYPI_DEPENDENCY_MAX_CONCURRENT`: Maximum concurrent dependency queries (default: 10) +- `PYPI_ENABLE_SECURITY_ANALYSIS`: Enable security vulnerability analysis (default: false) + +#### Example Configuration +```bash +# Use multiple mirror sources for better availability +export PYPI_INDEX_URL="https://pypi.org/pypi" +export PYPI_INDEX_URLS="https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/" +export PYPI_EXTRA_INDEX_URLS="https://test.pypi.org/simple/" + +# Private repository configuration +export PYPI_PRIVATE_PYPI_URL="https://private.pypi.company.com" +export PYPI_PRIVATE_PYPI_USERNAME="your_username" +export PYPI_PRIVATE_PYPI_PASSWORD="your_password" +``` ## Available MCP Tools diff --git a/examples/claude_desktop_config.json b/examples/claude_desktop_config.json index ddf6120..8eac9da 100644 --- a/examples/claude_desktop_config.json +++ b/examples/claude_desktop_config.json @@ -4,8 +4,14 @@ "command": "uvx", "args": ["--from", "pypi-query-mcp-server", "pypi-query-mcp"], "env": { - "PYPI_INDEX_URL": "https://pypi.org/simple/", - "CACHE_TTL": "3600" + "PYPI_INDEX_URL": "https://pypi.org/pypi", + "PYPI_INDEX_URLS": "https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/", + "PYPI_EXTRA_INDEX_URLS": "https://test.pypi.org/simple/", + "PYPI_CACHE_TTL": "3600", + "PYPI_LOG_LEVEL": "INFO", + "PYPI_REQUEST_TIMEOUT": "30.0", + "PYPI_DEPENDENCY_MAX_DEPTH": "5", + "PYPI_DEPENDENCY_MAX_CONCURRENT": "10" } } } diff --git a/examples/cursor_mcp.json b/examples/cursor_mcp.json index ddf6120..8eac9da 100644 --- a/examples/cursor_mcp.json +++ b/examples/cursor_mcp.json @@ -4,8 +4,14 @@ "command": "uvx", "args": ["--from", "pypi-query-mcp-server", "pypi-query-mcp"], "env": { - "PYPI_INDEX_URL": "https://pypi.org/simple/", - "CACHE_TTL": "3600" + "PYPI_INDEX_URL": "https://pypi.org/pypi", + "PYPI_INDEX_URLS": "https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/", + "PYPI_EXTRA_INDEX_URLS": "https://test.pypi.org/simple/", + "PYPI_CACHE_TTL": "3600", + "PYPI_LOG_LEVEL": "INFO", + "PYPI_REQUEST_TIMEOUT": "30.0", + "PYPI_DEPENDENCY_MAX_DEPTH": "5", + "PYPI_DEPENDENCY_MAX_CONCURRENT": "10" } } } diff --git a/examples/multi_mirror_demo.py b/examples/multi_mirror_demo.py new file mode 100644 index 0000000..6a5472c --- /dev/null +++ b/examples/multi_mirror_demo.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Demo script showing multi-mirror source configuration.""" + +import asyncio +import os + +from pypi_query_mcp.config import get_repository_manager, get_settings + + +async def demo_multi_mirror_configuration(): + """Demonstrate multi-mirror source configuration.""" + print("šŸ”§ PyPI Query MCP Server - Multi-Mirror Configuration Demo") + print("=" * 60) + + # Set up environment variables for demo + os.environ.update( + { + "PYPI_INDEX_URL": "https://pypi.org/pypi", + "PYPI_INDEX_URLS": "https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/", + "PYPI_EXTRA_INDEX_URLS": "https://test.pypi.org/simple/", + "PYPI_PRIVATE_PYPI_URL": "https://private.company.com/pypi", + "PYPI_PRIVATE_PYPI_USERNAME": "demo_user", + "PYPI_PRIVATE_PYPI_PASSWORD": "demo_password", + "PYPI_CACHE_TTL": "7200", + "PYPI_LOG_LEVEL": "DEBUG", + } + ) + + # Load settings + settings = get_settings() + + print("\nšŸ“‹ Configuration Settings:") + print(f" Primary Index URL: {settings.index_url}") + print(f" Cache TTL: {settings.cache_ttl} seconds") + print(f" Log Level: {settings.log_level}") + print(f" Request Timeout: {settings.request_timeout} seconds") + + print("\nšŸŒ Index URLs Configuration:") + all_urls = settings.get_all_index_urls() + primary_urls = settings.get_primary_index_urls() + fallback_urls = settings.get_fallback_index_urls() + + print(f" All Index URLs ({len(all_urls)}):") + for i, url in enumerate(all_urls, 1): + print(f" {i}. {url}") + + print(f"\n Primary URLs ({len(primary_urls)}):") + for i, url in enumerate(primary_urls, 1): + print(f" {i}. {url}") + + print(f"\n Fallback URLs ({len(fallback_urls)}):") + for i, url in enumerate(fallback_urls, 1): + print(f" {i}. {url}") + + print("\nšŸ” Private Repository Configuration:") + print(f" Has Private Repo: {settings.has_private_repo()}") + print(f" Has Private Auth: {settings.has_private_auth()}") + if settings.has_private_repo(): + print(f" Private URL: {settings.private_pypi_url}") + print(f" Username: {settings.private_pypi_username}") + print(f" Password: {'***' if settings.private_pypi_password else 'None'}") + + print("\nāš™ļø Advanced Settings:") + print(f" Dependency Max Depth: {settings.dependency_max_depth}") + print(f" Dependency Max Concurrent: {settings.dependency_max_concurrent}") + print(f" Security Analysis: {settings.enable_security_analysis}") + + # Load repository manager + repo_manager = get_repository_manager() + repo_manager.load_repositories_from_settings(settings) + + print("\nšŸ“¦ Repository Manager Configuration:") + all_repos = repo_manager.list_repositories() + enabled_repos = repo_manager.get_enabled_repositories() + private_repos = repo_manager.get_private_repositories() + + print(f" Total Repositories: {len(all_repos)}") + print(f" Enabled Repositories: {len(enabled_repos)}") + print(f" Private Repositories: {len(private_repos)}") + print(f" Has Private Repos: {repo_manager.has_private_repositories()}") + + print("\nšŸ“‹ Repository Details (by priority):") + for repo in enabled_repos: + auth_info = f" (Auth: {repo.auth_type.value})" if repo.requires_auth() else "" + print( + f" {repo.priority:3d}. {repo.name:12s} - {repo.type.value:7s} - {repo.url}{auth_info}" + ) + + print("\nāœ… Configuration loaded successfully!") + print("\nThis configuration provides:") + print(" ā€¢ High availability through multiple mirror sources") + print(" ā€¢ Automatic fallback to backup mirrors") + print(" ā€¢ Private repository support with authentication") + print(" ā€¢ Configurable dependency analysis settings") + print(" ā€¢ Secure credential management") + + +if __name__ == "__main__": + asyncio.run(demo_multi_mirror_configuration()) diff --git a/nox_actions/codetest.py b/nox_actions/codetest.py index 24f522a..87607a9 100644 --- a/nox_actions/codetest.py +++ b/nox_actions/codetest.py @@ -12,11 +12,14 @@ def pytest(session: nox.Session) -> None: session.install(".") session.install("pytest", "pytest-cov", "pytest-mock", "pytest-asyncio") test_root = os.path.join(THIS_ROOT, "tests") - session.run("pytest", f"--cov={PACKAGE_NAME}", - "--cov-report=xml:coverage.xml", - "--cov-report=term-missing", - f"--rootdir={test_root}", - env={"PYTHONPATH": THIS_ROOT.as_posix()}) + session.run( + "pytest", + f"--cov={PACKAGE_NAME}", + "--cov-report=xml:coverage.xml", + "--cov-report=term-missing", + f"--rootdir={test_root}", + env={"PYTHONPATH": THIS_ROOT.as_posix()}, + ) def mypy(session: nox.Session) -> None: diff --git a/nox_actions/lint.py b/nox_actions/lint.py index bec2943..1c48d79 100644 --- a/nox_actions/lint.py +++ b/nox_actions/lint.py @@ -15,4 +15,9 @@ def lint_fix(session: nox.Session) -> None: session.run("ruff", "check", "--fix") session.run("isort", ".") session.run("pre-commit", "run", "--all-files") - session.run("autoflake", "--in-place", "--remove-all-unused-imports", "--remove-unused-variables") + session.run( + "autoflake", + "--in-place", + "--remove-all-unused-imports", + "--remove-unused-variables", + ) diff --git a/nox_actions/release.py b/nox_actions/release.py index d736804..28119b6 100644 --- a/nox_actions/release.py +++ b/nox_actions/release.py @@ -21,7 +21,9 @@ def build(session: nox.Session) -> None: def build_exe(session: nox.Session) -> None: parser = argparse.ArgumentParser(prog="nox -s build-exe --release") parser.add_argument("--release", action="store_true") - parser.add_argument("--version", default="0.5.0", help="Version to use for the zip file") + parser.add_argument( + "--version", default="0.5.0", help="Version to use for the zip file" + ) parser.add_argument("--test", action="store_true") args = parser.parse_args(session.posargs) build_root = THIS_ROOT / "build" @@ -42,11 +44,17 @@ def build_exe(session: nox.Session) -> None: version = str(args.version) print(f"make zip to current version: {version}") os.makedirs(temp_dir, exist_ok=True) - zip_file = os.path.join(temp_dir, f"{PACKAGE_NAME}-{version}-{platform_name}.zip") + zip_file = os.path.join( + temp_dir, f"{PACKAGE_NAME}-{version}-{platform_name}.zip" + ) with zipfile.ZipFile(zip_file, "w") as zip_obj: for root, _, files in os.walk(platform_dir): for file in files: - zip_obj.write(os.path.join(root, file), - os.path.relpath(os.path.join(root, file), - os.path.join(platform_dir, "."))) + zip_obj.write( + os.path.join(root, file), + os.path.relpath( + os.path.join(root, file), + os.path.join(platform_dir, "."), + ), + ) print(f"Saving to {zip_file}") diff --git a/poetry.lock b/poetry.lock index 7b2e7db..0bfba97 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1474,4 +1474,4 @@ files = [ [metadata] lock-version = "2.1" python-versions = "^3.10" -content-hash = "1aa944c1b3c2cf066d5453f5cc675a4e90f7a283935839d91910ea3ff62e91cd" +content-hash = "25256463cd5ad4b2cc3ff5541fda1dd404d26dfa74d03569a7aa96607f39f3d4" diff --git a/pypi_query_mcp/config/__init__.py b/pypi_query_mcp/config/__init__.py index ad213e7..020f4f4 100644 --- a/pypi_query_mcp/config/__init__.py +++ b/pypi_query_mcp/config/__init__.py @@ -4,5 +4,32 @@ This package handles configuration loading, validation, and management for the MCP server, including private registry settings. """ -# Configuration exports will be added as modules are implemented -__all__ = [] +from .repository import ( + AuthType, + RepositoryConfig, + RepositoryManager, + RepositoryType, + get_repository_manager, + reload_repository_manager, +) +from .settings import ( + ServerSettings, + get_settings, + reload_settings, + update_settings, +) + +__all__ = [ + # Settings + "ServerSettings", + "get_settings", + "reload_settings", + "update_settings", + # Repository + "RepositoryConfig", + "RepositoryManager", + "RepositoryType", + "AuthType", + "get_repository_manager", + "reload_repository_manager", +] diff --git a/pypi_query_mcp/config/repository.py b/pypi_query_mcp/config/repository.py new file mode 100644 index 0000000..da2d381 --- /dev/null +++ b/pypi_query_mcp/config/repository.py @@ -0,0 +1,252 @@ +"""Repository configuration for PyPI Query MCP Server.""" + +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field, field_validator + + +class RepositoryType(str, Enum): + """Repository type enumeration.""" + + PUBLIC = "public" + PRIVATE = "private" + + +class AuthType(str, Enum): + """Authentication type enumeration.""" + + NONE = "none" + BASIC = "basic" + TOKEN = "token" + + +class RepositoryConfig(BaseModel): + """Configuration for a PyPI repository.""" + + name: str = Field(description="Repository name") + url: str = Field(description="Repository URL") + type: RepositoryType = Field(description="Repository type") + priority: int = Field( + default=100, description="Repository priority (lower = higher priority)" + ) + + # Authentication settings + auth_type: AuthType = Field( + default=AuthType.NONE, description="Authentication type" + ) + username: str | None = Field( + default=None, description="Username for authentication" + ) + password: str | None = Field( + default=None, description="Password for authentication" + ) + token: str | None = Field(default=None, description="Token for authentication") + + # Connection settings + timeout: float = Field(default=30.0, description="Request timeout in seconds") + max_retries: int = Field(default=3, description="Maximum retry attempts") + verify_ssl: bool = Field(default=True, description="Verify SSL certificates") + + # Feature flags + enabled: bool = Field(default=True, description="Whether repository is enabled") + use_cache: bool = Field(default=True, description="Whether to cache responses") + + @field_validator("priority") + @classmethod + def validate_priority(cls, v: int) -> int: + """Validate repository priority.""" + if v < 1 or v > 1000: + raise ValueError("Priority must be between 1 and 1000") + return v + + @field_validator("timeout") + @classmethod + def validate_timeout(cls, v: float) -> float: + """Validate timeout.""" + if v <= 0: + raise ValueError("Timeout must be positive") + return v + + @field_validator("max_retries") + @classmethod + def validate_max_retries(cls, v: int) -> int: + """Validate max retries.""" + if v < 0 or v > 10: + raise ValueError("Max retries must be between 0 and 10") + return v + + def requires_auth(self) -> bool: + """Check if repository requires authentication.""" + return self.auth_type != AuthType.NONE + + def has_credentials(self) -> bool: + """Check if repository has valid credentials.""" + if self.auth_type == AuthType.BASIC: + return bool(self.username and self.password) + elif self.auth_type == AuthType.TOKEN: + return bool(self.token) + return True # No auth required + + def get_safe_dict(self) -> dict[str, Any]: + """Get repository config as dictionary with sensitive data masked.""" + data = self.model_dump() + # Mask sensitive information + if data.get("password"): + data["password"] = "***" + if data.get("token"): + data["token"] = "***" + return data + + +class RepositoryManager: + """Manager for repository configurations.""" + + def __init__(self): + """Initialize repository manager.""" + self._repositories: dict[str, RepositoryConfig] = {} + self._load_default_repositories() + + def _load_default_repositories(self) -> None: + """Load default repository configurations.""" + # Add default public PyPI repository + public_repo = RepositoryConfig( + name="pypi", + url="https://pypi.org/pypi", + type=RepositoryType.PUBLIC, + priority=100, + auth_type=AuthType.NONE, + ) + self._repositories["pypi"] = public_repo + + def load_repositories_from_settings(self, settings) -> None: + """Load repositories from settings configuration.""" + # Clear existing repositories except default PyPI + repos_to_keep = { + name: repo + for name, repo in self._repositories.items() + if repo.type == RepositoryType.PUBLIC and name == "pypi" + } + self._repositories = repos_to_keep + + # Add repositories from index URLs + all_urls = settings.get_all_index_urls() + primary_urls = settings.get_primary_index_urls() + fallback_urls = settings.get_fallback_index_urls() + + # Update primary PyPI URL if different from default + if all_urls and all_urls[0] != "https://pypi.org/pypi": + self._repositories["pypi"].url = all_urls[0] + + # Add additional primary index URLs + for i, url in enumerate( + primary_urls[1:], 1 + ): # Skip first URL (already set as primary) + repo_name = f"index_{i}" + repo = RepositoryConfig( + name=repo_name, + url=url, + type=RepositoryType.PUBLIC, + priority=100 + i, # Slightly lower priority than primary + auth_type=AuthType.NONE, + ) + self._repositories[repo_name] = repo + + # Add fallback index URLs + for i, url in enumerate(fallback_urls): + repo_name = f"fallback_{i}" + repo = RepositoryConfig( + name=repo_name, + url=url, + type=RepositoryType.PUBLIC, + priority=200 + i, # Lower priority for fallbacks + auth_type=AuthType.NONE, + ) + self._repositories[repo_name] = repo + + # Add private repository if configured + if settings.has_private_repo(): + self.add_private_repository_from_settings( + settings.private_pypi_url, + settings.private_pypi_username, + settings.private_pypi_password, + ) + + def add_repository(self, repo: RepositoryConfig) -> None: + """Add a repository configuration.""" + if not repo.has_credentials() and repo.requires_auth(): + raise ValueError( + f"Repository {repo.name} requires authentication but has no credentials" + ) + self._repositories[repo.name] = repo + + def remove_repository(self, name: str) -> None: + """Remove a repository configuration.""" + if name == "pypi": + raise ValueError("Cannot remove default PyPI repository") + self._repositories.pop(name, None) + + def get_repository(self, name: str) -> RepositoryConfig | None: + """Get repository configuration by name.""" + return self._repositories.get(name) + + def list_repositories(self) -> list[RepositoryConfig]: + """List all repository configurations.""" + return list(self._repositories.values()) + + def get_enabled_repositories(self) -> list[RepositoryConfig]: + """Get all enabled repositories sorted by priority.""" + enabled = [repo for repo in self._repositories.values() if repo.enabled] + return sorted(enabled, key=lambda x: x.priority) + + def get_private_repositories(self) -> list[RepositoryConfig]: + """Get all private repositories.""" + return [ + repo + for repo in self._repositories.values() + if repo.type == RepositoryType.PRIVATE and repo.enabled + ] + + def has_private_repositories(self) -> bool: + """Check if any private repositories are configured.""" + return len(self.get_private_repositories()) > 0 + + def add_private_repository_from_settings( + self, url: str, username: str | None = None, password: str | None = None + ) -> None: + """Add private repository from settings.""" + if not url: + return + + auth_type = AuthType.BASIC if username and password else AuthType.NONE + + private_repo = RepositoryConfig( + name="private", + url=url, + type=RepositoryType.PRIVATE, + priority=1, # Higher priority than public + auth_type=auth_type, + username=username, + password=password, + ) + + self.add_repository(private_repo) + + +# Global repository manager instance +_repository_manager: RepositoryManager | None = None + + +def get_repository_manager() -> RepositoryManager: + """Get global repository manager instance.""" + global _repository_manager + if _repository_manager is None: + _repository_manager = RepositoryManager() + return _repository_manager + + +def reload_repository_manager() -> RepositoryManager: + """Reload repository manager.""" + global _repository_manager + _repository_manager = RepositoryManager() + return _repository_manager diff --git a/pypi_query_mcp/config/settings.py b/pypi_query_mcp/config/settings.py new file mode 100644 index 0000000..4e3d68c --- /dev/null +++ b/pypi_query_mcp/config/settings.py @@ -0,0 +1,198 @@ +"""Configuration settings for PyPI Query MCP Server.""" + +from typing import Any + +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class ServerSettings(BaseSettings): + """Server configuration settings.""" + + model_config = SettingsConfigDict( + env_prefix="PYPI_", + env_file=".env", + env_file_encoding="utf-8", + case_sensitive=False, + extra="ignore", + ) + + # Basic server settings + log_level: str = Field(default="INFO", description="Logging level") + cache_ttl: int = Field(default=3600, description="Cache time-to-live in seconds") + request_timeout: float = Field( + default=30.0, description="HTTP request timeout in seconds" + ) + max_retries: int = Field(default=3, description="Maximum number of retry attempts") + retry_delay: float = Field( + default=1.0, description="Delay between retries in seconds" + ) + + # PyPI settings + index_url: str = Field( + default="https://pypi.org/pypi", description="Primary PyPI index URL" + ) + index_urls: str | None = Field( + default=None, description="Additional PyPI index URLs (comma-separated)" + ) + extra_index_urls: str | None = Field( + default=None, description="Extra PyPI index URLs for fallback (comma-separated)" + ) + + # Private repository settings + private_pypi_url: str | None = Field( + default=None, description="Private PyPI repository URL" + ) + private_pypi_username: str | None = Field( + default=None, description="Private PyPI username" + ) + private_pypi_password: str | None = Field( + default=None, description="Private PyPI password" + ) + + # Advanced dependency analysis settings + dependency_max_depth: int = Field( + default=5, description="Maximum depth for recursive dependency analysis" + ) + dependency_max_concurrent: int = Field( + default=10, description="Maximum concurrent dependency queries" + ) + enable_security_analysis: bool = Field( + default=False, description="Enable security vulnerability analysis" + ) + + @field_validator("log_level") + @classmethod + def validate_log_level(cls, v: str) -> str: + """Validate log level.""" + valid_levels = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} + if v.upper() not in valid_levels: + raise ValueError(f"Invalid log level: {v}. Must be one of {valid_levels}") + return v.upper() + + @field_validator("cache_ttl") + @classmethod + def validate_cache_ttl(cls, v: int) -> int: + """Validate cache TTL.""" + if v < 0: + raise ValueError("Cache TTL must be non-negative") + return v + + @field_validator("dependency_max_depth") + @classmethod + def validate_dependency_max_depth(cls, v: int) -> int: + """Validate dependency analysis max depth.""" + if v < 1 or v > 10: + raise ValueError("Dependency max depth must be between 1 and 10") + return v + + @field_validator("dependency_max_concurrent") + @classmethod + def validate_dependency_max_concurrent(cls, v: int) -> int: + """Validate max concurrent dependency queries.""" + if v < 1 or v > 50: + raise ValueError("Max concurrent queries must be between 1 and 50") + return v + + def has_private_repo(self) -> bool: + """Check if private repository is configured.""" + return bool(self.private_pypi_url) + + def has_private_auth(self) -> bool: + """Check if private repository authentication is configured.""" + return bool( + self.private_pypi_url + and self.private_pypi_username + and self.private_pypi_password + ) + + def get_all_index_urls(self) -> list[str]: + """Get all configured index URLs in priority order.""" + urls = [self.index_url] + + # Add additional index URLs + if self.index_urls: + additional_urls = [ + url.strip() for url in self.index_urls.split(",") if url.strip() + ] + urls.extend(additional_urls) + + # Add extra index URLs (lower priority) + if self.extra_index_urls: + extra_urls = [ + url.strip() for url in self.extra_index_urls.split(",") if url.strip() + ] + urls.extend(extra_urls) + + # Remove duplicates while preserving order + seen = set() + unique_urls = [] + for url in urls: + if url not in seen: + seen.add(url) + unique_urls.append(url) + + return unique_urls + + def get_primary_index_urls(self) -> list[str]: + """Get primary index URLs (excluding extra fallback URLs).""" + urls = [self.index_url] + + if self.index_urls: + additional_urls = [ + url.strip() for url in self.index_urls.split(",") if url.strip() + ] + urls.extend(additional_urls) + + # Remove duplicates while preserving order + seen = set() + unique_urls = [] + for url in urls: + if url not in seen: + seen.add(url) + unique_urls.append(url) + + return unique_urls + + def get_fallback_index_urls(self) -> list[str]: + """Get fallback index URLs.""" + if not self.extra_index_urls: + return [] + + return [url.strip() for url in self.extra_index_urls.split(",") if url.strip()] + + def get_safe_dict(self) -> dict[str, Any]: + """Get configuration as dictionary with sensitive data masked.""" + data = self.model_dump() + # Mask sensitive information + if data.get("private_pypi_password"): + data["private_pypi_password"] = "***" + return data + + +# Global settings instance +_settings: ServerSettings | None = None + + +def get_settings() -> ServerSettings: + """Get global settings instance.""" + global _settings + if _settings is None: + _settings = ServerSettings() + return _settings + + +def reload_settings() -> ServerSettings: + """Reload settings from environment variables.""" + global _settings + _settings = ServerSettings() + return _settings + + +def update_settings(**kwargs: Any) -> ServerSettings: + """Update settings with new values.""" + global _settings + current_data = _settings.model_dump() if _settings else {} + current_data.update(kwargs) + _settings = ServerSettings(**current_data) + return _settings diff --git a/pypi_query_mcp/core/pypi_client.py b/pypi_query_mcp/core/pypi_client.py index 5e02425..c8bf944 100644 --- a/pypi_query_mcp/core/pypi_client.py +++ b/pypi_query_mcp/core/pypi_client.py @@ -99,6 +99,7 @@ class PyPIClient: def _is_cache_valid(self, cache_entry: dict[str, Any]) -> bool: """Check if cache entry is still valid.""" import time + return time.time() - cache_entry.get("timestamp", 0) < self._cache_ttl async def _make_request(self, url: str) -> dict[str, Any]: @@ -140,7 +141,7 @@ class PyPIClient: else: raise PyPIServerError( response.status_code, - f"Unexpected status code: {response.status_code}" + f"Unexpected status code: {response.status_code}", ) except httpx.TimeoutException as e: @@ -155,12 +156,16 @@ class PyPIClient: # Wait before retry (except on last attempt) if attempt < self.max_retries: - await asyncio.sleep(self.retry_delay * (2 ** attempt)) # Exponential backoff + await asyncio.sleep( + self.retry_delay * (2**attempt) + ) # Exponential backoff # If we get here, all retries failed raise last_exception - async def get_package_info(self, package_name: str, use_cache: bool = True) -> dict[str, Any]: + async def get_package_info( + self, package_name: str, use_cache: bool = True + ) -> dict[str, Any]: """Get comprehensive package information from PyPI. Args: @@ -194,10 +199,8 @@ class PyPIClient: # Cache the result import time - self._cache[cache_key] = { - "data": data, - "timestamp": time.time() - } + + self._cache[cache_key] = {"data": data, "timestamp": time.time()} return data @@ -205,7 +208,9 @@ class PyPIClient: logger.error(f"Failed to fetch package info for {normalized_name}: {e}") raise - async def get_package_versions(self, package_name: str, use_cache: bool = True) -> list[str]: + async def get_package_versions( + self, package_name: str, use_cache: bool = True + ) -> list[str]: """Get list of available versions for a package. Args: @@ -219,7 +224,9 @@ class PyPIClient: releases = package_info.get("releases", {}) return list(releases.keys()) - async def get_latest_version(self, package_name: str, use_cache: bool = True) -> str: + async def get_latest_version( + self, package_name: str, use_cache: bool = True + ) -> str: """Get the latest version of a package. Args: diff --git a/pypi_query_mcp/core/version_utils.py b/pypi_query_mcp/core/version_utils.py index 9e327e4..610afcb 100644 --- a/pypi_query_mcp/core/version_utils.py +++ b/pypi_query_mcp/core/version_utils.py @@ -45,7 +45,9 @@ class VersionCompatibility: logger.warning(f"Failed to parse requires_python '{requires_python}': {e}") return None - def extract_python_versions_from_classifiers(self, classifiers: list[str]) -> set[str]: + def extract_python_versions_from_classifiers( + self, classifiers: list[str] + ) -> set[str]: """Extract Python version information from classifiers. Args: @@ -87,7 +89,7 @@ class VersionCompatibility: self, target_version: str, requires_python: str | None = None, - classifiers: list[str] | None = None + classifiers: list[str] | None = None, ) -> dict[str, Any]: """Check if a target Python version is compatible with package requirements. @@ -105,7 +107,7 @@ class VersionCompatibility: "compatibility_source": None, "details": {}, "warnings": [], - "suggestions": [] + "suggestions": [], } try: @@ -119,15 +121,17 @@ class VersionCompatibility: spec_set = self.parse_requires_python(requires_python) if spec_set: is_compatible = target_ver in spec_set - result.update({ - "is_compatible": is_compatible, - "compatibility_source": "requires_python", - "details": { - "requires_python": requires_python, - "parsed_spec": str(spec_set), - "check_result": is_compatible + result.update( + { + "is_compatible": is_compatible, + "compatibility_source": "requires_python", + "details": { + "requires_python": requires_python, + "parsed_spec": str(spec_set), + "check_result": is_compatible, + }, } - }) + ) if not is_compatible: result["suggestions"].append( @@ -139,7 +143,9 @@ class VersionCompatibility: # Fall back to classifiers if no requires_python if classifiers: - supported_versions = self.extract_python_versions_from_classifiers(classifiers) + supported_versions = self.extract_python_versions_from_classifiers( + classifiers + ) implementations = self.extract_python_implementations(classifiers) if supported_versions: @@ -148,21 +154,23 @@ class VersionCompatibility: target_major = str(target_ver.major) is_compatible = ( - target_version in supported_versions or - target_major_minor in supported_versions or - target_major in supported_versions + target_version in supported_versions + or target_major_minor in supported_versions + or target_major in supported_versions ) - result.update({ - "is_compatible": is_compatible, - "compatibility_source": "classifiers", - "details": { - "supported_versions": sorted(supported_versions), - "implementations": sorted(implementations), - "target_major_minor": target_major_minor, - "check_result": is_compatible + result.update( + { + "is_compatible": is_compatible, + "compatibility_source": "classifiers", + "details": { + "supported_versions": sorted(supported_versions), + "implementations": sorted(implementations), + "target_major_minor": target_major_minor, + "check_result": is_compatible, + }, } - }) + ) if not is_compatible: result["suggestions"].append( @@ -186,7 +194,7 @@ class VersionCompatibility: self, requires_python: str | None = None, classifiers: list[str] | None = None, - available_pythons: list[str] | None = None + available_pythons: list[str] | None = None, ) -> dict[str, Any]: """Get list of compatible Python versions for a package. @@ -200,9 +208,7 @@ class VersionCompatibility: """ if available_pythons is None: # Default Python versions to check - available_pythons = [ - "3.7", "3.8", "3.9", "3.10", "3.11", "3.12", "3.13" - ] + available_pythons = ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12", "3.13"] compatible = [] incompatible = [] @@ -213,28 +219,34 @@ class VersionCompatibility: ) if result["is_compatible"]: - compatible.append({ - "version": python_version, - "source": result["compatibility_source"] - }) + compatible.append( + { + "version": python_version, + "source": result["compatibility_source"], + } + ) else: - incompatible.append({ - "version": python_version, - "reason": result["suggestions"][0] if result["suggestions"] else "Unknown" - }) + incompatible.append( + { + "version": python_version, + "reason": result["suggestions"][0] + if result["suggestions"] + else "Unknown", + } + ) return { "compatible_versions": compatible, "incompatible_versions": incompatible, "total_checked": len(available_pythons), - "compatibility_rate": len(compatible) / len(available_pythons) if available_pythons else 0, - "recommendations": self._generate_recommendations(compatible, incompatible) + "compatibility_rate": len(compatible) / len(available_pythons) + if available_pythons + else 0, + "recommendations": self._generate_recommendations(compatible, incompatible), } def _generate_recommendations( - self, - compatible: list[dict[str, Any]], - incompatible: list[dict[str, Any]] + self, compatible: list[dict[str, Any]], incompatible: list[dict[str, Any]] ) -> list[str]: """Generate recommendations based on compatibility results. diff --git a/pypi_query_mcp/server.py b/pypi_query_mcp/server.py index c3a53c0..898e396 100644 --- a/pypi_query_mcp/server.py +++ b/pypi_query_mcp/server.py @@ -17,8 +17,7 @@ from .tools import ( # Configure logging logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) @@ -60,14 +59,14 @@ async def get_package_info(package_name: str) -> dict[str, Any]: return { "error": str(e), "error_type": type(e).__name__, - "package_name": package_name + "package_name": package_name, } except Exception as e: logger.error(f"Unexpected error querying package {package_name}: {e}") return { "error": f"Unexpected error: {e}", "error_type": "UnexpectedError", - "package_name": package_name + "package_name": package_name, } @@ -103,19 +102,21 @@ async def get_package_versions(package_name: str) -> dict[str, Any]: return { "error": str(e), "error_type": type(e).__name__, - "package_name": package_name + "package_name": package_name, } except Exception as e: logger.error(f"Unexpected error querying versions for {package_name}: {e}") return { "error": f"Unexpected error: {e}", "error_type": "UnexpectedError", - "package_name": package_name + "package_name": package_name, } @mcp.tool() -async def get_package_dependencies(package_name: str, version: str | None = None) -> dict[str, Any]: +async def get_package_dependencies( + package_name: str, version: str | None = None +) -> dict[str, Any]: """Get dependency information for a PyPI package. This tool retrieves comprehensive dependency information for a Python package, @@ -138,8 +139,10 @@ async def get_package_dependencies(package_name: str, version: str | None = None NetworkError: For network-related errors """ try: - logger.info(f"MCP tool: Querying dependencies for {package_name}" + - (f" version {version}" if version else " (latest)")) + logger.info( + f"MCP tool: Querying dependencies for {package_name}" + + (f" version {version}" if version else " (latest)") + ) result = await query_package_dependencies(package_name, version) logger.info(f"Successfully retrieved dependencies for package: {package_name}") return result @@ -149,7 +152,7 @@ async def get_package_dependencies(package_name: str, version: str | None = None "error": str(e), "error_type": type(e).__name__, "package_name": package_name, - "version": version + "version": version, } except Exception as e: logger.error(f"Unexpected error querying dependencies for {package_name}: {e}") @@ -157,15 +160,13 @@ async def get_package_dependencies(package_name: str, version: str | None = None "error": f"Unexpected error: {e}", "error_type": "UnexpectedError", "package_name": package_name, - "version": version + "version": version, } @mcp.tool() async def check_package_python_compatibility( - package_name: str, - target_python_version: str, - use_cache: bool = True + package_name: str, target_python_version: str, use_cache: bool = True ) -> dict[str, Any]: """Check if a package is compatible with a specific Python version. @@ -190,8 +191,12 @@ async def check_package_python_compatibility( NetworkError: For network-related errors """ try: - logger.info(f"MCP tool: Checking Python {target_python_version} compatibility for {package_name}") - result = await check_python_compatibility(package_name, target_python_version, use_cache) + logger.info( + f"MCP tool: Checking Python {target_python_version} compatibility for {package_name}" + ) + result = await check_python_compatibility( + package_name, target_python_version, use_cache + ) logger.info(f"Compatibility check completed for {package_name}") return result except (InvalidPackageNameError, PackageNotFoundError, NetworkError) as e: @@ -200,7 +205,7 @@ async def check_package_python_compatibility( "error": str(e), "error_type": type(e).__name__, "package_name": package_name, - "target_python_version": target_python_version + "target_python_version": target_python_version, } except Exception as e: logger.error(f"Unexpected error checking compatibility for {package_name}: {e}") @@ -208,15 +213,13 @@ async def check_package_python_compatibility( "error": f"Unexpected error: {e}", "error_type": "UnexpectedError", "package_name": package_name, - "target_python_version": target_python_version + "target_python_version": target_python_version, } @mcp.tool() async def get_package_compatible_python_versions( - package_name: str, - python_versions: list[str] | None = None, - use_cache: bool = True + package_name: str, python_versions: list[str] | None = None, use_cache: bool = True ) -> dict[str, Any]: """Get all Python versions compatible with a package. @@ -242,7 +245,9 @@ async def get_package_compatible_python_versions( """ try: logger.info(f"MCP tool: Getting compatible Python versions for {package_name}") - result = await get_compatible_python_versions(package_name, python_versions, use_cache) + result = await get_compatible_python_versions( + package_name, python_versions, use_cache + ) logger.info(f"Compatible versions analysis completed for {package_name}") return result except (InvalidPackageNameError, PackageNotFoundError, NetworkError) as e: @@ -250,14 +255,16 @@ async def get_package_compatible_python_versions( return { "error": str(e), "error_type": type(e).__name__, - "package_name": package_name + "package_name": package_name, } except Exception as e: - logger.error(f"Unexpected error getting compatible versions for {package_name}: {e}") + logger.error( + f"Unexpected error getting compatible versions for {package_name}: {e}" + ) return { "error": f"Unexpected error: {e}", "error_type": "UnexpectedError", - "package_name": package_name + "package_name": package_name, } @@ -266,7 +273,7 @@ async def get_package_compatible_python_versions( "--log-level", default="INFO", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]), - help="Logging level" + help="Logging level", ) def main(log_level: str) -> None: """Start the PyPI Query MCP Server.""" diff --git a/pypi_query_mcp/tools/compatibility_check.py b/pypi_query_mcp/tools/compatibility_check.py index 20896c7..b559035 100644 --- a/pypi_query_mcp/tools/compatibility_check.py +++ b/pypi_query_mcp/tools/compatibility_check.py @@ -10,9 +10,7 @@ logger = logging.getLogger(__name__) async def check_python_compatibility( - package_name: str, - target_python_version: str, - use_cache: bool = True + package_name: str, target_python_version: str, use_cache: bool = True ) -> dict[str, Any]: """Check if a package is compatible with a specific Python version. @@ -35,7 +33,9 @@ async def check_python_compatibility( if not target_python_version or not target_python_version.strip(): raise ValueError("Target Python version cannot be empty") - logger.info(f"Checking Python {target_python_version} compatibility for package: {package_name}") + logger.info( + f"Checking Python {target_python_version} compatibility for package: {package_name}" + ) try: async with PyPIClient() as client: @@ -48,19 +48,25 @@ async def check_python_compatibility( # Perform compatibility check compat_checker = VersionCompatibility() result = compat_checker.check_version_compatibility( - target_python_version, - requires_python, - classifiers + target_python_version, requires_python, classifiers ) # Add package information to result - result.update({ - "package_name": info.get("name", package_name), - "package_version": info.get("version", ""), - "requires_python": requires_python, - "supported_implementations": compat_checker.extract_python_implementations(classifiers), - "classifier_versions": sorted(compat_checker.extract_python_versions_from_classifiers(classifiers)) - }) + result.update( + { + "package_name": info.get("name", package_name), + "package_version": info.get("version", ""), + "requires_python": requires_python, + "supported_implementations": compat_checker.extract_python_implementations( + classifiers + ), + "classifier_versions": sorted( + compat_checker.extract_python_versions_from_classifiers( + classifiers + ) + ), + } + ) return result @@ -73,9 +79,7 @@ async def check_python_compatibility( async def get_compatible_python_versions( - package_name: str, - python_versions: list[str] | None = None, - use_cache: bool = True + package_name: str, python_versions: list[str] | None = None, use_cache: bool = True ) -> dict[str, Any]: """Get list of Python versions compatible with a package. @@ -108,19 +112,25 @@ async def get_compatible_python_versions( # Get compatibility information compat_checker = VersionCompatibility() result = compat_checker.get_compatible_versions( - requires_python, - classifiers, - python_versions + requires_python, classifiers, python_versions ) # Add package information to result - result.update({ - "package_name": info.get("name", package_name), - "package_version": info.get("version", ""), - "requires_python": requires_python, - "supported_implementations": sorted(compat_checker.extract_python_implementations(classifiers)), - "classifier_versions": sorted(compat_checker.extract_python_versions_from_classifiers(classifiers)) - }) + result.update( + { + "package_name": info.get("name", package_name), + "package_version": info.get("version", ""), + "requires_python": requires_python, + "supported_implementations": sorted( + compat_checker.extract_python_implementations(classifiers) + ), + "classifier_versions": sorted( + compat_checker.extract_python_versions_from_classifiers( + classifiers + ) + ), + } + ) return result @@ -128,13 +138,14 @@ async def get_compatible_python_versions( # Re-raise PyPI-specific errors raise except Exception as e: - logger.error(f"Unexpected error getting compatible versions for {package_name}: {e}") + logger.error( + f"Unexpected error getting compatible versions for {package_name}: {e}" + ) raise NetworkError(f"Failed to get compatible Python versions: {e}", e) from e async def suggest_python_version_for_packages( - package_names: list[str], - use_cache: bool = True + package_names: list[str], use_cache: bool = True ) -> dict[str, Any]: """Suggest optimal Python version for a list of packages. @@ -152,7 +163,9 @@ async def suggest_python_version_for_packages( if not package_names: raise ValueError("Package names list cannot be empty") - logger.info(f"Analyzing Python version compatibility for {len(package_names)} packages") + logger.info( + f"Analyzing Python version compatibility for {len(package_names)} packages" + ) # Default Python versions to analyze python_versions = ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"] @@ -172,20 +185,20 @@ async def suggest_python_version_for_packages( compat_checker = VersionCompatibility() compat_result = compat_checker.get_compatible_versions( - requires_python, - classifiers, - python_versions + requires_python, classifiers, python_versions ) # Store compatibility for this package - compatible_versions = [v["version"] for v in compat_result["compatible_versions"]] + compatible_versions = [ + v["version"] for v in compat_result["compatible_versions"] + ] compatibility_matrix[package_name] = compatible_versions package_details[package_name] = { "version": info.get("version", ""), "requires_python": requires_python, "compatible_versions": compatible_versions, - "compatibility_rate": compat_result["compatibility_rate"] + "compatibility_rate": compat_result["compatibility_rate"], } except Exception as e: @@ -207,14 +220,18 @@ async def suggest_python_version_for_packages( # Generate recommendations recommendations = [] if common_versions: - latest_common = max(common_versions, key=lambda x: tuple(map(int, x.split(".")))) + latest_common = max( + common_versions, key=lambda x: tuple(map(int, x.split("."))) + ) recommendations.append( f"āœ… Recommended Python version: {latest_common} " f"(compatible with all {len([p for p in compatibility_matrix if compatibility_matrix[p]])} packages)" ) if len(common_versions) > 1: - all_common = sorted(common_versions, key=lambda x: tuple(map(int, x.split(".")))) + all_common = sorted( + common_versions, key=lambda x: tuple(map(int, x.split("."))) + ) recommendations.append( f"šŸ“‹ All compatible versions: {', '.join(all_common)}" ) @@ -227,13 +244,19 @@ async def suggest_python_version_for_packages( # Find the version compatible with most packages version_scores = {} for version in python_versions: - score = sum(1 for compatible in compatibility_matrix.values() if version in compatible) + score = sum( + 1 + for compatible in compatibility_matrix.values() + if version in compatible + ) version_scores[version] = score if version_scores: best_version = max(version_scores, key=version_scores.get) best_score = version_scores[best_version] - total_packages = len([p for p in compatibility_matrix if compatibility_matrix[p]]) + total_packages = len( + [p for p in compatibility_matrix if compatibility_matrix[p]] + ) if best_score > 0: recommendations.append( @@ -246,10 +269,14 @@ async def suggest_python_version_for_packages( "successful_analyses": len(package_details), "failed_analyses": len(errors), "common_compatible_versions": sorted(common_versions), - "recommended_version": max(common_versions, key=lambda x: tuple(map(int, x.split(".")))) if common_versions else None, + "recommended_version": max( + common_versions, key=lambda x: tuple(map(int, x.split("."))) + ) + if common_versions + else None, "compatibility_matrix": compatibility_matrix, "package_details": package_details, "errors": errors, "recommendations": recommendations, - "python_versions_analyzed": python_versions + "python_versions_analyzed": python_versions, } diff --git a/pypi_query_mcp/tools/package_query.py b/pypi_query_mcp/tools/package_query.py index cf36a0f..b8eb2ba 100644 --- a/pypi_query_mcp/tools/package_query.py +++ b/pypi_query_mcp/tools/package_query.py @@ -24,7 +24,9 @@ def format_package_info(package_data: dict[str, Any]) -> dict[str, Any]: "name": info.get("name", ""), "version": info.get("version", ""), "summary": info.get("summary", ""), - "description": info.get("description", "")[:500] + "..." if len(info.get("description", "")) > 500 else info.get("description", ""), + "description": info.get("description", "")[:500] + "..." + if len(info.get("description", "")) > 500 + else info.get("description", ""), "author": info.get("author", ""), "author_email": info.get("author_email", ""), "maintainer": info.get("maintainer", ""), @@ -53,7 +55,13 @@ def format_package_info(package_data: dict[str, Any]) -> dict[str, Any]: formatted["download_info"] = { "files_count": len(urls), "file_types": list({url.get("packagetype", "") for url in urls}), - "python_versions": list({url.get("python_version", "") for url in urls if url.get("python_version")}), + "python_versions": list( + { + url.get("python_version", "") + for url in urls + if url.get("python_version") + } + ), } return formatted @@ -83,11 +91,16 @@ def format_version_info(package_data: dict[str, Any]) -> dict[str, Any]: "version_details": { version: { "release_count": len(releases[version]), - "has_wheel": any(file.get("packagetype") == "bdist_wheel" for file in releases[version]), - "has_source": any(file.get("packagetype") == "sdist" for file in releases[version]), + "has_wheel": any( + file.get("packagetype") == "bdist_wheel" + for file in releases[version] + ), + "has_source": any( + file.get("packagetype") == "sdist" for file in releases[version] + ), } for version in sorted_versions[:10] # Details for last 10 versions - } + }, } @@ -120,7 +133,7 @@ def format_dependency_info(package_data: dict[str, Any]) -> dict[str, Any]: extra_part = parts[1] if len(parts) > 1 else "" if "extra ==" in extra_part: - extra_name = extra_part.split("extra ==")[1].strip().strip('"\'') + extra_name = extra_part.split("extra ==")[1].strip().strip("\"'") if extra_name not in optional_deps: optional_deps[extra_name] = [] optional_deps[extra_name].append(dep_name) @@ -142,7 +155,7 @@ def format_dependency_info(package_data: dict[str, Any]) -> dict[str, Any]: "dev_count": len(dev_deps), "optional_groups": len(optional_deps), "total_optional": sum(len(deps) for deps in optional_deps.values()), - } + }, } @@ -208,7 +221,9 @@ async def query_package_versions(package_name: str) -> dict[str, Any]: raise NetworkError(f"Failed to query package versions: {e}", e) from e -async def query_package_dependencies(package_name: str, version: str | None = None) -> dict[str, Any]: +async def query_package_dependencies( + package_name: str, version: str | None = None +) -> dict[str, Any]: """Query package dependency information from PyPI. Args: @@ -226,8 +241,10 @@ async def query_package_dependencies(package_name: str, version: str | None = No if not package_name or not package_name.strip(): raise InvalidPackageNameError(package_name) - logger.info(f"Querying dependencies for package: {package_name}" + - (f" version {version}" if version else " (latest)")) + logger.info( + f"Querying dependencies for package: {package_name}" + + (f" version {version}" if version else " (latest)") + ) try: async with PyPIClient() as client: @@ -236,8 +253,10 @@ async def query_package_dependencies(package_name: str, version: str | None = No # TODO: In future, support querying specific version dependencies # For now, we return dependencies for the latest version if version and version != package_data.get("info", {}).get("version"): - logger.warning(f"Specific version {version} requested but not implemented yet. " - f"Returning dependencies for latest version.") + logger.warning( + f"Specific version {version} requested but not implemented yet. " + f"Returning dependencies for latest version." + ) return format_dependency_info(package_data) except PyPIError: diff --git a/pyproject.toml b/pyproject.toml index 2f57f0c..660e6de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ fastmcp = "^0.4.0" httpx = "^0.28.0" packaging = "^24.0" pydantic = "^2.0.0" +pydantic-settings = "^2.0.0" click = "^8.1.0" [tool.poetry.group.dev.dependencies] diff --git a/tests/conftest.py b/tests/conftest.py index 59cdba9..797a70a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -19,7 +19,7 @@ def sample_package_data(): "requires_dist": [ "requests>=2.25.0", "click>=8.0.0", - "pytest>=6.0.0; extra == 'test'" + "pytest>=6.0.0; extra == 'test'", ], "classifiers": [ "Development Status :: 4 - Beta", @@ -30,30 +30,30 @@ def sample_package_data(): "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", - "Programming Language :: Python :: Implementation :: CPython" - ] + "Programming Language :: Python :: Implementation :: CPython", + ], }, "releases": { "1.0.0": [ { "filename": "test_package-1.0.0-py3-none-any.whl", "packagetype": "bdist_wheel", - "python_version": "py3" + "python_version": "py3", }, { "filename": "test-package-1.0.0.tar.gz", "packagetype": "sdist", - "python_version": "source" - } + "python_version": "source", + }, ], "0.9.0": [ { "filename": "test-package-0.9.0.tar.gz", "packagetype": "sdist", - "python_version": "source" + "python_version": "source", } - ] - } + ], + }, } diff --git a/tests/test_basic.py b/tests/test_basic.py index 8a392c4..3617f50 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -47,7 +47,7 @@ def test_version_compatibility(): classifiers = [ "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", - "Programming Language :: Python :: Implementation :: CPython" + "Programming Language :: Python :: Implementation :: CPython", ] versions = compat.extract_python_versions_from_classifiers(classifiers) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..ecdc224 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,386 @@ +"""Tests for configuration management.""" + +import os +from unittest.mock import patch + +import pytest + +from pypi_query_mcp.config import ( + AuthType, + RepositoryConfig, + RepositoryManager, + RepositoryType, + ServerSettings, + get_repository_manager, + get_settings, + reload_settings, +) + + +class TestServerSettings: + """Test ServerSettings class.""" + + def test_default_settings(self): + """Test default settings values.""" + settings = ServerSettings() + + assert settings.log_level == "INFO" + assert settings.cache_ttl == 3600 + assert settings.request_timeout == 30.0 + assert settings.max_retries == 3 + assert settings.retry_delay == 1.0 + assert settings.index_url == "https://pypi.org/pypi" + assert settings.private_pypi_url is None + assert settings.dependency_max_depth == 5 + assert settings.dependency_max_concurrent == 10 + assert settings.enable_security_analysis is False + + def test_environment_variables(self): + """Test loading from environment variables.""" + with patch.dict( + os.environ, + { + "PYPI_LOG_LEVEL": "DEBUG", + "PYPI_CACHE_TTL": "7200", + "PYPI_PRIVATE_PYPI_URL": "https://private.pypi.com", + "PYPI_PRIVATE_PYPI_USERNAME": "testuser", + "PYPI_PRIVATE_PYPI_PASSWORD": "testpass", + }, + ): + settings = ServerSettings() + + assert settings.log_level == "DEBUG" + assert settings.cache_ttl == 7200 + assert settings.private_pypi_url == "https://private.pypi.com" + assert settings.private_pypi_username == "testuser" + assert settings.private_pypi_password == "testpass" + + def test_validation(self): + """Test settings validation.""" + # Test invalid log level + with pytest.raises(ValueError, match="Invalid log level"): + ServerSettings(log_level="INVALID") + + # Test negative cache TTL + with pytest.raises(ValueError, match="Cache TTL must be non-negative"): + ServerSettings(cache_ttl=-1) + + # Test invalid dependency max depth + with pytest.raises( + ValueError, match="Dependency max depth must be between 1 and 10" + ): + ServerSettings(dependency_max_depth=0) + + with pytest.raises( + ValueError, match="Dependency max depth must be between 1 and 10" + ): + ServerSettings(dependency_max_depth=11) + + def test_has_private_repo(self): + """Test private repository detection.""" + settings = ServerSettings() + assert not settings.has_private_repo() + + settings = ServerSettings(private_pypi_url="https://private.pypi.com") + assert settings.has_private_repo() + + def test_has_private_auth(self): + """Test private authentication detection.""" + settings = ServerSettings() + assert not settings.has_private_auth() + + settings = ServerSettings( + private_pypi_url="https://private.pypi.com", + private_pypi_username="user", + private_pypi_password="pass", + ) + assert settings.has_private_auth() + + def test_get_safe_dict(self): + """Test safe dictionary representation.""" + settings = ServerSettings(private_pypi_password="secret123") + safe_dict = settings.get_safe_dict() + assert safe_dict["private_pypi_password"] == "***" + + def test_multiple_index_urls(self): + """Test multiple index URLs configuration.""" + settings = ServerSettings( + index_url="https://pypi.org/pypi", + index_urls="https://mirrors.aliyun.com/pypi/simple/,https://pypi.tuna.tsinghua.edu.cn/simple/", + extra_index_urls="https://test.pypi.org/simple/", + ) + + all_urls = settings.get_all_index_urls() + assert len(all_urls) == 4 + assert all_urls[0] == "https://pypi.org/pypi" + assert "https://mirrors.aliyun.com/pypi/simple/" in all_urls + assert "https://pypi.tuna.tsinghua.edu.cn/simple/" in all_urls + assert "https://test.pypi.org/simple/" in all_urls + + primary_urls = settings.get_primary_index_urls() + assert len(primary_urls) == 3 + assert "https://test.pypi.org/simple/" not in primary_urls + + fallback_urls = settings.get_fallback_index_urls() + assert len(fallback_urls) == 1 + assert fallback_urls[0] == "https://test.pypi.org/simple/" + + def test_duplicate_urls_removal(self): + """Test duplicate URL removal while preserving order.""" + settings = ServerSettings( + index_url="https://pypi.org/pypi", + index_urls="https://pypi.org/pypi,https://mirrors.aliyun.com/pypi/simple/", + extra_index_urls="https://mirrors.aliyun.com/pypi/simple/", + ) + + all_urls = settings.get_all_index_urls() + assert len(all_urls) == 2 # Duplicates removed + assert all_urls[0] == "https://pypi.org/pypi" + assert all_urls[1] == "https://mirrors.aliyun.com/pypi/simple/" + + def test_empty_index_urls(self): + """Test handling of empty index URLs.""" + settings = ServerSettings( + index_url="https://pypi.org/pypi", index_urls="", extra_index_urls=None + ) + + all_urls = settings.get_all_index_urls() + assert len(all_urls) == 1 + assert all_urls[0] == "https://pypi.org/pypi" + + fallback_urls = settings.get_fallback_index_urls() + assert len(fallback_urls) == 0 + + +class TestRepositoryConfig: + """Test RepositoryConfig class.""" + + def test_basic_repository(self): + """Test basic repository configuration.""" + repo = RepositoryConfig( + name="test", url="https://test.pypi.com", type=RepositoryType.PRIVATE + ) + + assert repo.name == "test" + assert repo.url == "https://test.pypi.com" + assert repo.type == RepositoryType.PRIVATE + assert repo.priority == 100 + assert repo.auth_type == AuthType.NONE + assert repo.enabled is True + + def test_repository_with_auth(self): + """Test repository with authentication.""" + repo = RepositoryConfig( + name="private", + url="https://private.pypi.com", + type=RepositoryType.PRIVATE, + auth_type=AuthType.BASIC, + username="user", + password="pass", + ) + + assert repo.requires_auth() + assert repo.has_credentials() + + def test_repository_validation(self): + """Test repository validation.""" + # Test invalid priority + with pytest.raises(ValueError, match="Priority must be between 1 and 1000"): + RepositoryConfig( + name="test", + url="https://test.com", + type=RepositoryType.PUBLIC, + priority=0, + ) + + # Test invalid timeout + with pytest.raises(ValueError, match="Timeout must be positive"): + RepositoryConfig( + name="test", + url="https://test.com", + type=RepositoryType.PUBLIC, + timeout=0, + ) + + def test_get_safe_dict(self): + """Test safe dictionary representation.""" + repo = RepositoryConfig( + name="test", + url="https://test.com", + type=RepositoryType.PRIVATE, + auth_type=AuthType.BASIC, + username="user", + password="secret123", + ) + safe_dict = repo.get_safe_dict() + assert safe_dict["password"] == "***" + + +class TestRepositoryManager: + """Test RepositoryManager class.""" + + def test_default_repositories(self): + """Test default repository loading.""" + manager = RepositoryManager() + repos = manager.list_repositories() + + assert len(repos) == 1 + assert repos[0].name == "pypi" + assert repos[0].type == RepositoryType.PUBLIC + + def test_add_repository(self): + """Test adding repository.""" + manager = RepositoryManager() + + repo = RepositoryConfig( + name="private", + url="https://private.pypi.com", + type=RepositoryType.PRIVATE, + priority=1, + ) + + manager.add_repository(repo) + assert len(manager.list_repositories()) == 2 + assert manager.get_repository("private") == repo + + def test_remove_repository(self): + """Test removing repository.""" + manager = RepositoryManager() + + # Cannot remove default PyPI + with pytest.raises(ValueError, match="Cannot remove default PyPI repository"): + manager.remove_repository("pypi") + + # Add and remove custom repository + repo = RepositoryConfig( + name="test", url="https://test.com", type=RepositoryType.PRIVATE + ) + manager.add_repository(repo) + assert manager.get_repository("test") is not None + + manager.remove_repository("test") + assert manager.get_repository("test") is None + + def test_get_enabled_repositories(self): + """Test getting enabled repositories sorted by priority.""" + manager = RepositoryManager() + + # Add repositories with different priorities + repo1 = RepositoryConfig( + name="high_priority", + url="https://high.com", + type=RepositoryType.PRIVATE, + priority=1, + ) + repo2 = RepositoryConfig( + name="low_priority", + url="https://low.com", + type=RepositoryType.PRIVATE, + priority=200, + ) + + manager.add_repository(repo1) + manager.add_repository(repo2) + + enabled = manager.get_enabled_repositories() + assert len(enabled) == 3 # Including default PyPI + assert enabled[0].name == "high_priority" # Highest priority first + assert enabled[1].name == "pypi" + assert enabled[2].name == "low_priority" + + def test_private_repositories(self): + """Test private repository management.""" + manager = RepositoryManager() + + assert not manager.has_private_repositories() + assert len(manager.get_private_repositories()) == 0 + + # Add private repository + repo = RepositoryConfig( + name="private", url="https://private.com", type=RepositoryType.PRIVATE + ) + manager.add_repository(repo) + + assert manager.has_private_repositories() + assert len(manager.get_private_repositories()) == 1 + + def test_add_private_repository_from_settings(self): + """Test adding private repository from settings.""" + manager = RepositoryManager() + + # Add without auth + manager.add_private_repository_from_settings("https://private.com") + private_repos = manager.get_private_repositories() + assert len(private_repos) == 1 + assert private_repos[0].auth_type == AuthType.NONE + + # Add with auth + manager = RepositoryManager() + manager.add_private_repository_from_settings( + "https://private.com", "user", "pass" + ) + private_repos = manager.get_private_repositories() + assert len(private_repos) == 1 + assert private_repos[0].auth_type == AuthType.BASIC + assert private_repos[0].username == "user" + + def test_load_repositories_from_settings(self): + """Test loading repositories from settings.""" + manager = RepositoryManager() + + # Create settings with multiple index URLs + settings = ServerSettings( + index_url="https://custom.pypi.org/pypi", + index_urls="https://mirrors.aliyun.com/pypi/simple/", + extra_index_urls="https://test.pypi.org/simple/", + private_pypi_url="https://private.pypi.com", + private_pypi_username="user", + private_pypi_password="pass", + ) + + manager.load_repositories_from_settings(settings) + + # Check all repositories are loaded + all_repos = manager.list_repositories() + repo_names = [repo.name for repo in all_repos] + + assert "pypi" in repo_names # Primary + assert "index_1" in repo_names # Additional index + assert "fallback_0" in repo_names # Fallback + assert "private" in repo_names # Private repo + + # Check primary PyPI URL is updated + pypi_repo = manager.get_repository("pypi") + assert pypi_repo.url == "https://custom.pypi.org/pypi" + + # Check priorities are correct + enabled_repos = manager.get_enabled_repositories() + priorities = [repo.priority for repo in enabled_repos] + assert priorities == sorted(priorities) # Should be sorted by priority + + # Check private repository has auth + private_repo = manager.get_repository("private") + assert private_repo.auth_type == AuthType.BASIC + assert private_repo.username == "user" + + +class TestGlobalInstances: + """Test global configuration instances.""" + + def test_get_settings(self): + """Test global settings instance.""" + settings1 = get_settings() + settings2 = get_settings() + assert settings1 is settings2 # Same instance + + def test_reload_settings(self): + """Test settings reload.""" + settings1 = get_settings() + settings2 = reload_settings() + assert settings1 is not settings2 # Different instance + + def test_get_repository_manager(self): + """Test global repository manager instance.""" + manager1 = get_repository_manager() + manager2 = get_repository_manager() + assert manager1 is manager2 # Same instance