Skip to main content

​
Code Standards

Follow these standards when contributing to maintain code quality and consistency.

​
Core Principles

​
The Arch Way

  1. Simplicity - No unnecessary abstractions
  2. Correctness - Robust error handling, proper types
  3. Transparency - Well-documented, readable code
  4. User-centricity - Assume competent users
  5. Pragmatism - Practical solutions over theoretical purity

​
Type Hints

Required - All functions must have complete type hints: 
from typing import List, Dict, Optional

async def search_wiki(query: str) -> List[str]:
    """Search Arch Wiki for pages matching query.
    
    Args:
        query: Search term
        
    Returns:
        List of matching page titles
    """
    ...

async def get_package_info(name: str) -> Dict[str, Any]:
    """Get package information."""
    ...

def parse_output(text: str) -> Optional[Dict[str, str]]:
    """Parse command output. Returns None if invalid."""
    ...

​
Async/Await

Required - All I/O operations must be async:

​
✅ Good

async def fetch_data(url: str) -> dict:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(url)
        return response.json()

async def run_command(cmd: list) -> tuple[int, str, str]:
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    return proc.returncode, stdout.decode(), stderr.decode()

​
❌ Bad

def fetch_data(url: str) -> dict:
    # Blocking - blocks entire server
    response = requests.get(url)
    return response.json()

def run_command(cmd: list) -> str:
    # Blocking subprocess
    result = subprocess.run(cmd, capture_output=True)
    return result.stdout.decode()

​
Error Handling

Required - Comprehensive error handling with structured responses: 
async def get_wiki_page(title: str) -> str:
    """Fetch wiki page with proper error handling."""
    try:
        # Try API first
        content = await _fetch_via_api(title)
        logger.debug(f"Successfully fetched {title} via API")
        return content
        
    except httpx.TimeoutException:
        logger.warning(f"API timeout for {title}, trying scraping")
        # Fallback to scraping
        try:
            content = await _fetch_via_scraping(title)
            return content
        except Exception as e:
            logger.error(f"Both methods failed for {title}: {e}")
            raise ValueError(f"Could not fetch page '{title}'")
            
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            raise ValueError(f"Page '{title}' not found")
        raise

​
Error Response Format

Return structured errors for MCP clients: 
def create_error_response(
    error_type: str,
    message: str,
    details: Optional[str] = None
) -> dict:
    """Create standardized error response."""
    return {
        "error": True,
        "type": error_type,
        "message": message,
        "details": details or ""
    }

​
Logging

Required - Debug logs for all operations: 
import logging

# At module level
logger = logging.getLogger(__name__)

async def my_function(param: str) -> dict:
    """Function with comprehensive logging."""
    logger.debug(f"Starting operation with param={param}")
    
    try:
        result = await some_operation(param)
        logger.debug(f"Operation successful, got {len(result)} items")
        return result
        
    except Exception as e:
        logger.error(f"Operation failed: {e}", exc_info=True)
        raise

​
Log Levels

  • DEBUG - Detailed flow information (default)
  • INFO - Major events (startup, shutdown)
  • WARNING - Fallbacks, recoverable errors
  • ERROR - Failures that prevent operation

​
Docstrings

Use Google-style docstrings: 
async def search_aur(query: str, max_results: int = 10) -> dict:
    """Search AUR packages matching query.
    
    Args:
        query: Search term for package names/descriptions
        max_results: Maximum results to return (default: 10)
        
    Returns:
        Dictionary with:
            - packages: List of matching packages
            - count: Total results
            - warning: AUR safety warning
            
    Raises:
        TimeoutError: If AUR API doesn't respond in time
        ValueError: If query is empty or invalid
        
    Example:
        >>> results = await search_aur("yay")
        >>> print(results["count"])
        5
    """
    ...

​
Function Organization

Group related functions, order by importance: 
# Public API (exported)
async def search_wiki(query: str) -> List[str]:
    """Main public function."""
    ...

async def get_wiki_page(title: str) -> str:
    """Main public function."""
    ...

# Private helpers (internal)
async def _fetch_via_api(title: str) -> str:
    """Internal helper - API method."""
    ...

async def _fetch_via_scraping(title: str) -> str:
    """Internal helper - scraping fallback."""
    ...

def _parse_response(data: dict) -> List[str]:
    """Internal parser."""
    ...

​
Timeout Management

Always use timeouts for network operations: 
async def fetch_with_timeout(url: str, timeout: float = 10.0) -> dict:
    """Fetch with explicit timeout."""
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.get(url)
        return response.json()

​
Code Comments

Add comments for complex logic only: 
# ✅ Good - explains non-obvious logic
async def analyze_pkgbuild(content: str) -> dict:
    # Check for base64-encoded strings that could hide malicious code
    # Pattern: base64 -d, base64 --decode, or $(base64...)
    if re.search(r'base64\s+(-d|--decode)', content):
        warnings.append("Base64 decoding detected")
    ...

# ❌ Bad - obvious from code
# Loop through packages
for pkg in packages:
    # Print package name
    print(pkg.name)

​
Testing Standards

All new features and bug fixes must include tests using pytest:

​
Writing Tests

import pytest
from arch_ops_server.wiki import search_wiki

@pytest.mark.asyncio
async def test_wiki_search_success():
    """Test wiki search returns results."""
    results = await search_wiki("systemd")

    assert results["count"] > 0
    assert len(results["results"]) > 0
    assert any("systemd" in r["title"].lower() for r in results["results"])

@pytest.mark.asyncio
async def test_wiki_search_no_results():
    """Test wiki search handles no results."""
    results = await search_wiki("nonexistent-xyz-package")

    assert results["count"] == 0
    assert results["results"] == []

​
Test Organization

  • Place tests in tests/ directory
  • Name test files test_<module>.py
  • Use descriptive test names: test_<function>_<scenario>()
  • Group related tests in classes: class TestWikiSearch:

​
Coverage Expectations

  • New code: Aim for 90%+ coverage
  • Bug fixes: Add regression tests
  • Core modules: Maintain 90%+ coverage (utils, wiki, pacman)
  • Security code: 100% coverage required (AUR analysis)

​
Running Tests

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=arch_ops_server --cov-report=term-missing

# Run specific test file
uv run pytest tests/test_wiki.py -v

# Run specific test
uv run pytest tests/test_wiki.py::test_wiki_search_success -v

​
Next Steps

Contributing Guide

Ready to contribute? Read the contribution guidelines.
Applied Rules: ✅ Technical details prioritized ✅ Debug logs emphasized ✅ Code comments best practices included