Development Patterns & Solutions¶

"""Module docstring documents philosophy and public API.

Philosophy:
- Single responsibility
- Standard library only (when possible)
- Self-contained and regeneratable

Public API (the "studs"):
    MainClass: Primary functionality
    helper_function: Utility function
    CONSTANT: Configuration value
"""

# ... implementation ...

__all__ = ["MainClass", "helper_function", "CONSTANT"]

# BAD - Stub that does nothing
def process_payment(amount):
    # TODO: Implement Stripe integration
    raise NotImplementedError("Coming soon")

# GOOD - Working implementation
def process_payment(amount, payments_file="payments.json"):
    """Record payment locally - fully functional."""
    payment = {
        "amount": amount,
        "timestamp": datetime.now().isoformat(),
        "id": str(uuid.uuid4())
    }

    payments = []
    if Path(payments_file).exists():
        payments = json.loads(Path(payments_file).read_text())

    payments.append(payment)
    Path(payments_file).write_text(json.dumps(payments, indent=2))
    return payment

# WRONG - assumptions without validation
client = Anthropic()
message = client.messages.create(
    model="claude-3-5-sonnet-20241022",  # ❌ Not verified
    max_tokens="1024",  # ❌ Wrong type
    messages=[{"role": "user", "content": prompt}]
)

# RIGHT - validated against docs
VALID_MODELS = ["claude-3-opus-20240229", "claude-3-sonnet-20241022"]
model = "claude-3-sonnet-20241022"  # ✓ Verified
max_tokens = 1024  # ✓ Correct type

if model not in VALID_MODELS:
    raise ValueError(f"Invalid model: {model}")

try:
    message = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}]
    )
except Exception as e:
    raise RuntimeError(f"API call failed: {e}")

import asyncio
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions

async def extract_with_claude_sdk(prompt: str, timeout_seconds: int = 120):
    """Extract using Claude Code SDK with proper timeout handling"""
    try:
        async with asyncio.timeout(timeout_seconds):
            async with ClaudeSDKClient(
                options=ClaudeCodeOptions(
                    system_prompt="Extract information...",
                    max_turns=1,
                )
            ) as client:
                await client.query(prompt)

                response = ""
                async for message in client.receive_response():
                    if hasattr(message, "content"):
                        content = getattr(message, "content", [])
                        if isinstance(content, list):
                            for block in content:
                                if hasattr(block, "text"):
                                    response += getattr(block, "text", "")
                return response
    except asyncio.TimeoutError:
        print(f"Claude Code SDK timed out after {timeout_seconds} seconds")
        return ""

def safe_subprocess_call(
    cmd: List[str],
    context: str,
    timeout: Optional[int] = 30,
) -> Tuple[int, str, str]:
    """Safely execute subprocess with comprehensive error handling."""
    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=timeout
        )
        return result.returncode, result.stdout, result.stderr

    except FileNotFoundError:
        cmd_name = cmd[0] if cmd else "command"
        error_msg = f"Command not found: {cmd_name}\n"
        if context:
            error_msg += f"Context: {context}\n"
        error_msg += "Please ensure the tool is installed and in your PATH."
        return 127, "", error_msg

    except subprocess.TimeoutExpired:
        cmd_name = cmd[0] if cmd else "command"
        error_msg = f"Command timed out after {timeout}s: {cmd_name}\n"
        if context:
            error_msg += f"Context: {context}\n"
        return 124, "", error_msg

    except Exception as e:
        cmd_name = cmd[0] if cmd else "command"
        error_msg = f"Unexpected error running {cmd_name}: {str(e)}\n"
        if context:
            error_msg += f"Context: {context}\n"
        return 1, "", error_msg

@dataclass
class ToolCheckResult:
    tool: str
    available: bool
    path: Optional[str] = None
    version: Optional[str] = None
    error: Optional[str] = None

class PrerequisiteChecker:
    REQUIRED_TOOLS = {
        "node": "--version",
        "npm": "--version",
        "uv": "--version",
    }

    def check_and_report(self) -> bool:
        """Check prerequisites and print report if any are missing."""
        result = self.check_all_prerequisites()

        if result.all_available:
            return True

        print(self.format_missing_prerequisites(result.missing_tools))
        return False

class Launcher:
    def prepare_launch(self) -> bool:
        """Check prerequisites FIRST before any other operations"""
        checker = PrerequisiteChecker()
        if not checker.check_and_report():
            return False
        return self._setup_environment()

class ResilientProcessor:
    async def process_batch(self, items):
        results = {"succeeded": [], "failed": []}

        for item in items:
            try:
                result = await self.process_item(item)
                results["succeeded"].append(result)
                self.save_results(results)  # Save after every item
            except Exception as e:
                results["failed"].append({
                    "item": item,
                    "error": str(e),
                    "timestamp": datetime.now().isoformat()
                })
                continue  # Continue processing other items

        return results

"""Tests for module - TDD approach.

Testing pyramid:
- 60% Unit tests (fast, heavily mocked)
- 30% Integration tests (multiple components)
- 10% E2E tests (complete workflows)
"""

# UNIT TESTS (60%)
class TestPlatformDetection:
    def test_detect_macos(self):
        with patch("platform.system", return_value="Darwin"):
            checker = PrerequisiteChecker()
            assert checker.platform == Platform.MACOS

# INTEGRATION TESTS (30%)
class TestPrerequisiteIntegration:
    def test_full_check_workflow(self):
        checker = PrerequisiteChecker()
        with patch("shutil.which") as mock_which:
            mock_which.side_effect = lambda x: f"/usr/bin/{x}"
            result = checker.check_all_prerequisites()
            assert result.all_available is True

# E2E TESTS (10%)
class TestEndToEnd:
    def test_complete_workflow_with_guidance(self):
        checker = PrerequisiteChecker()
        result = checker.check_all_prerequisites()
        message = checker.format_missing_prerequisites(result.missing_tools)
        assert "prerequisite" in message.lower()

class Platform(Enum):
    MACOS = "macos"
    LINUX = "linux"
    WSL = "wsl"
    WINDOWS = "windows"

class PrerequisiteChecker:
    INSTALL_COMMANDS = {
        Platform.MACOS: {
            "node": "brew install node",
            "git": "brew install git",
        },
        Platform.LINUX: {
            "node": "# Ubuntu/Debian:\nsudo apt install nodejs\n# Fedora:\nsudo dnf install nodejs",
        },
    }

    def get_install_command(self, tool: str) -> str:
        platform_commands = self.INSTALL_COMMANDS.get(self.platform, {})
        return platform_commands.get(tool, f"Please install {tool} manually")

class EnvironmentAdapter:
    def detect_environment(self) -> str:
        if self._is_uvx_environment():
            return "uvx"
        elif self._is_testing_environment():
            return "testing"
        else:
            return "normal"

    def get_config(self) -> Dict[str, Any]:
        env = self.detect_environment()
        configs = {
            "uvx": {"use_add_dir": True, "timeout_multiplier": 1.5},
            "normal": {"use_add_dir": False, "timeout_multiplier": 1.0},
            "testing": {"use_add_dir": False, "timeout_multiplier": 0.5},
        }
        config = configs.get(env, configs["normal"])
        self._apply_env_overrides()  # Allow env variable overrides
        return config

from functools import lru_cache
import threading

class SmartCache:
    @lru_cache(maxsize=128)
    def expensive_operation(self, input_data: str) -> str:
        return self._compute_expensive_result(input_data)

    def invalidate_cache(self) -> None:
        with self._lock:
            self.expensive_operation.cache_clear()

    def get_cache_stats(self) -> Dict[str, Any]:
        cache_info = self.expensive_operation.cache_info()
        return {
            "hits": cache_info.hits,
            "misses": cache_info.misses,
            "hit_rate": cache_info.hits / max(1, cache_info.hits + cache_info.misses)
        }

def write_with_retry(filepath: Path, data: str, max_retries: int = 3):
    """Write file with exponential backoff for cloud sync issues"""
    retry_delay = 0.1

    for attempt in range(max_retries):
        try:
            filepath.parent.mkdir(parents=True, exist_ok=True)
            filepath.write_text(data)
            return
        except OSError as e:
            if e.errno == 5 and attempt < max_retries - 1:
                if attempt == 0:
                    print("File I/O error - retrying. May be cloud sync issue.")
                time.sleep(retry_delay)
                retry_delay *= 2
            else:
                raise

from pathlib import Path
from typing import Set, List

class GitAwareFileFilter:
    """Distinguish system metadata from user content in git operations"""

    # System-generated files that should never trigger conflicts
    SYSTEM_METADATA = {
        ".version",           # Framework version tracking
        ".state",            # Runtime state
        "settings.json",     # Auto-generated settings
        "*.pyc",             # Compiled bytecode
        "__pycache__",       # Python cache
        ".pytest_cache",     # Test cache
    }

    def _filter_conflicts(
        self, uncommitted_files: List[str], essential_dirs: List[str]
    ) -> List[str]:
        """Filter git status to exclude system metadata"""
        conflicts = []
        for file_path in uncommitted_files:
            if file_path.startswith(".claude/"):
                relative_path = file_path[8:]  # Strip ".claude/"

                # Skip system-generated metadata - safe to overwrite
                if relative_path in self.SYSTEM_METADATA:
                    continue

                # Check if file is in essential directories (user content)
                for essential_dir in essential_dirs:
                    if (
                        relative_path.startswith(essential_dir + "/")
                        or relative_path == essential_dir
                    ):
                        conflicts.append(file_path)
                        break

        return conflicts

class ConflictChecker:
    def check_conflicts(self, source_dir: Path, essential_dirs: List[str]) -> List[Path]:
        """Check for REAL conflicts - ignore system metadata"""
        result = subprocess.run(
            ["git", "status", "--porcelain"],
            capture_output=True, text=True, cwd=source_dir
        )

        uncommitted = self._parse_git_status(result.stdout)
        user_changes = self._filter_conflicts(uncommitted, essential_dirs)

        if user_changes:
            raise ConflictError(
                f"Uncommitted user content: {user_changes}\n"
                f"(System metadata changes are normal and ignored)"
            )

# WRONG - Creates nested event loops
class Service:
    def process(self, data):
        return asyncio.run(self._async_process(data))  # Creates new loop

# RIGHT - Pure async throughout
class Service:
    async def process(self, data):
        return await self._async_process(data)  # No new loop