API Reference¶

This document provides detailed API documentation for Zaojun. The API is designed for programmatic use and integration with other Python tools, including the new caching system for improved performance.

Overview¶

Zaojun provides a simple Python API for checking dependency versions. The main functionality is available through the check() function, but individual components can also be used separately.

Main Functions¶

`check()`¶

The main entry point for checking dependencies.

def check(
    pyproject_toml: Path = Path("./pyproject.toml"),
    short: bool = False,
    groups: bool = False,
    compat_ok: bool = False,
    cache: bool = False,
    clear_cache: bool = False,
    cache_stats: bool = False,
) -> None:
    """Check listed dependencies against latest versions available on Pypi.org.

    Args:
        pyproject_toml: Full path to pyproject.toml file to check.
        short: Should only the shortest possible output be produced.
        groups: Should dependencies in groups be checked.
        compat_ok: Compatible versions are considered OK and won't cause an exit code of 1.
        cache: Enable caching of PyPI API responses (default: False).
        clear_cache: Clear all cache entries before checking.
        cache_stats: Show cache statistics after checking.

    Returns:
        None, but exits with code 0 (success) or 1 (updates needed).

    Raises:
        FileNotFoundError: If pyproject_toml file doesn't exist.
        tomllib.TOMLDecodeError: If pyproject_toml contains invalid TOML.
        ValueError: If dependencies have invalid format.
    """

Example Usage:

from pathlib import Path
from zaojun import check

# Basic check
check()

# Check with all options
check(
    pyproject_toml=Path("/path/to/project/pyproject.toml"),
    short=True,
    groups=True,
    compat_ok=True,
    cache=True,
    clear_cache=False,
    cache_stats=True,
)

`process_dependencies()`¶

Process a list of dependencies and check each one.

def process_dependencies(
    dependencies: list[Any],
    short: bool,
    compat_ok: bool,
    client: httpx.Client,
    cache: PyPICache | None = None,
) -> bool:
    """Check all dependencies for updates.

    Args:
        dependencies: List of dependency strings to check.
        short: Should only the shortest possible output be produced.
        compat_ok: Compatible versions are considered OK.
        client: HTTP client for making requests to PyPI.
        cache: Optional PyPICache instance for caching PyPI responses.

    Returns:
        True if incompatible update is available or if compat_ok is False
        on any update available.
    """

Example Usage:

import httpx
from zaojun import process_dependencies

client = httpx.Client()
dependencies = ["httpx~=0.28.1", "packaging~=26.0"]

needs_update = process_dependencies(
    dependencies=dependencies,
    short=False,
    compat_ok=False,
    client=client,
    cache=None,
)

client.close()

`check_dependency()`¶

Check if a single dependency needs an update.

def check_dependency(
    dependency: str,
    short: bool,
    compat_ok: bool,
    client: httpx.Client,
    cache: PyPICache | None = None,
) -> bool:
    """Check if a dependency needs an update.

    Args:
        dependency: Dependency string to check.
        short: Should only the shortest possible output be produced.
        compat_ok: Compatible versions are considered OK.
        client: HTTP client for making requests to PyPI.
        cache: Optional PyPICache instance for caching PyPI responses.

    Returns:
        True if newer version of dependency is available and that version
        is outside version spec or compat_ok is set to false and newer
        version is within version spec.

    Raises:
        ValueError: If dependency has invalid format.
        httpx.HTTPStatusError: If PyPI API request fails.
        httpx.RequestError: If network error occurs.
    """

Example Usage:

import httpx
from zaojun import check_dependency

client = httpx.Client()

# Check a single dependency
needs_update = check_dependency(
    dependency="httpx~=0.28.1",
    short=False,
    compat_ok=False,
    client=client,
    cache=None,
)

client.close()

`parse_dependency()`¶

Parse a dependency string into its components.

def parse_dependency(dependency: str) -> tuple[str, str, str | None]:
    """Parse a dependency string into (package_name, version_constraint, environment_marker).

    Args:
        dependency: Dependency string to parse.

    Returns:
        Tuple containing:
        - package_name: Name of the package
        - version_constraint: Version specification string
        - environment_marker: Environment marker or None

    Raises:
        ValueError: If dependency has invalid format.
    """

Example Usage:

from zaojun import parse_dependency

# Parse a simple dependency
package_name, version_spec, marker = parse_dependency("httpx~=0.28.1")
# Returns: ("httpx", "~=0.28.1", None)

# Parse with environment marker
package_name, version_spec, marker = parse_dependency(
    "package>=1.0; python_version>='3.8'"
)
# Returns: ("package", ">=1.0", "python_version>='3.8'")

`is_version_compatible()`¶

Check if a version is compatible with constraints.

def is_version_compatible(current_spec: str, latest_version: str) -> bool:
    """Check if the latest version satisfies the current version constraints.

    Args:
        current_spec: Version constraint string.
        latest_version: Latest version to check.

    Returns:
        True if latest_version satisfies current_spec, False otherwise.
    """

Example Usage:

from zaojun import is_version_compatible

# Check compatibility
compatible = is_version_compatible("~=1.2.3", "1.2.5")
# Returns: True (1.2.5 satisfies ~=1.2.3)

compatible = is_version_compatible("==1.2.3", "1.2.5")
# Returns: False (1.2.5 doesn't satisfy ==1.2.3)

`is_latest_version()`¶

Check if the latest version has been specified.

def is_latest_version(current_spec: str, latest_version: str) -> bool:
    """Check if the latest version has been specified.

    Args:
        current_spec: Version constraint string.
        latest_version: Latest version to check.

    Returns:
        True if latest_version is explicitly specified in current_spec.
    """

Example Usage:

from zaojun import is_latest_version

# Check if latest is specified
is_latest = is_latest_version("==1.2.3", "1.2.3")
# Returns: True (1.2.3 is explicitly specified)

is_latest = is_latest_version(">=1.2.3", "1.2.5")
# Returns: False (1.2.5 is not explicitly specified)

`is_local_dependency()`¶

Check if a dependency is local or URL-based.

def is_local_dependency(dependency: str) -> bool:
    """Check if dependency is a local path or URL.

    Args:
        dependency: Dependency string to check.

    Returns:
        True if dependency is local or URL-based, False otherwise.
    """

Example Usage:

from zaojun import is_local_dependency

# Check local dependencies
is_local = is_local_dependency("./local-package")
# Returns: True

is_local = is_local_dependency("package @ https://example.com/pkg.zip")
# Returns: True

is_local = is_local_dependency("package==1.0.0")
# Returns: False

`get_latest_pypi_version()`¶

Get the latest version of a package from PyPI.

def get_latest_pypi_version(package_name: str, client: httpx.Client) -> str:
    """Get latest version number of a package on Pypi.org.

    Args:
        package_name: Name of the package to check.
        client: HTTP client for making requests to PyPI.

    Returns:
        Latest version string from PyPI.

    Raises:
        httpx.HTTPStatusError: If PyPI API request fails.
        httpx.RequestError: If network error occurs.
    """

Example Usage:

import httpx
from zaojun import get_latest_pypi_version

client = httpx.Client()

# Get latest version
latest_version = get_latest_pypi_version("httpx", client, cache=None)
# Returns: e.g., "0.28.1"

client.close()

CLI Interface¶

`main()`¶

The CLI entry point.

def main() -> None:
    """Run actual processing."""

Example Usage:

from zaojun import main

# This is equivalent to running `zaojun` from command line
if __name__ == "__main__":
    main()

`app`¶

The Cyclopts application instance.

app = App()

Example Usage:

from zaojun import app

# Use the app programmatically
if __name__ == "__main__":
    app()

Cache API¶

`PyPICache` Class¶

The PyPICache class provides caching functionality for PyPI API responses to reduce network calls and improve performance.

class PyPICache:
    """Cache for PyPI API responses with TTL-based expiration.

    Args:
        cache_dir: Directory to store cache files. If None, uses
                  platform-specific cache directory.
        ttl_hours: Time-to-live for cache entries in hours.
        enabled: Whether caching is enabled.
    """

    def __init__(
        self,
        cache_dir: Path | None = None,
        ttl_hours: int = 24,
        enabled: bool = True,
    ):
        """Initialize the PyPI cache."""

    def get(self, package_name: str) -> dict | None:
        """Get cached data for a package.

        Args:
            package_name: Name of the package.

        Returns:
            Cached data dictionary or None if not found/expired.
        """

    def set(self, package_name: str, data: dict) -> bool:
        """Cache data for a package.

        Args:
            package_name: Name of the package.
            data: Data to cache.

        Returns:
            True if successful, False otherwise.
        """

    def clear(self, package_name: str | None = None) -> int:
        """Clear cache entries.

        Args:
            package_name: If specified, clear only this package's cache.
                         If None, clear all cache entries.

        Returns:
            Number of cache entries cleared.
        """

    def get_stats(self) -> dict:
        """Get cache statistics.

        Returns:
            Dictionary with cache statistics.
        """

    def is_expired(self, package_name: str) -> bool:
        """Check if a cache entry exists and is expired.

        Args:
            package_name: Name of the package.

        Returns:
            True if cache entry exists and is expired, False otherwise.
        """

`get_default_cache()`¶

def get_default_cache() -> PyPICache:
    """Get the default cache instance.

    Returns:
        Default PyPICache instance.
    """

Example Usage:

from zaojun.cache import PyPICache, get_default_cache

# Get default cache
cache = get_default_cache()

# Or create custom cache
custom_cache = PyPICache(
    cache_dir=Path("/tmp/my-cache"),
    ttl_hours=12,
    enabled=True,
)

# Use cache with API functions
from zaojun import get_latest_pypi_version
import httpx

client = httpx.Client()
latest_version = get_latest_pypi_version("httpx", client, cache=cache)
client.close()

# Check cache statistics
stats = cache.get_stats()
print(f"Cache hits: {stats['hits']}, misses: {stats['misses']}")

# Clear cache
cleared = cache.clear()
print(f"Cleared {cleared} cache entries")

Types and Constants¶

Import Statements¶

from pathlib import Path
from typing import Annotated, Any
import httpx
import tomllib
from cyclopts import App, Parameter
from packaging.requirements import InvalidRequirement, Requirement
from packaging.specifiers import SpecifierSet
from packaging.version import InvalidVersion, parse as parse_version
from .cache import PyPICache

Type Aliases¶

The API uses standard Python types: - Path for file paths - bool for flags - str for strings - list[Any] for dependency lists - tuple[str, str, str | None] for parsed dependencies - dict for cache data and statistics - PyPICache | None for optional cache instances

Error Handling¶

Common Exceptions¶

ValueError: Raised for invalid dependency formats
FileNotFoundError: Raised when pyproject.toml doesn't exist
tomllib.TOMLDecodeError: Raised for invalid TOML
httpx.HTTPStatusError: Raised for HTTP errors from PyPI
httpx.RequestError: Raised for network errors

Error Recovery¶

import httpx
from zaojun import check_dependency

client = httpx.Client()

try:
    needs_update = check_dependency(
        dependency="invalid-package-format",
        short=False,
        compat_ok=False,
        client=client,
        cache=None,
    )
except ValueError as e:
    print(f"Invalid dependency: {e}")
except httpx.HTTPStatusError as e:
    print(f"PyPI API error: {e}")
except httpx.RequestError as e:
    print(f"Network error: {e}")
finally:
    client.close()

Cache-Specific Error Handling¶

from zaojun.cache import PyPICache

cache = PyPICache()

try:
    # Cache operations
    cache.set("package", {"version": "1.0.0"})
    data = cache.get("package")
    cleared = cache.clear()
except (OSError, json.JSONDecodeError) as e:
    print(f"Cache error: {e}")
    # Cache automatically handles corrupted files

Cache Performance Considerations¶

Cache Configuration¶

from zaojun.cache import PyPICache
from pathlib import Path

# Custom cache configuration
cache = PyPICache(
    cache_dir=Path.home() / ".myapp" / "cache",  # Custom location
    ttl_hours=48,  # Longer TTL for stable packages
    enabled=True,  # Enable caching
)

# Monitor cache effectiveness
stats = cache.get_stats()
if stats["hits"] > stats["misses"] * 2:
    print("Cache is effective (more hits than misses)")
else:
    print("Consider adjusting TTL or cache strategy")

Cache Integration Example¶

"""Advanced cache integration example."""

import httpx
from pathlib import Path
from zaojun.cache import PyPICache
from zaojun import get_latest_pypi_version

class CachedDependencyChecker:
    """Dependency checker with intelligent caching."""

    def __init__(self, cache_dir: Path | None = None):
        self.cache = PyPICache(cache_dir=cache_dir, ttl_hours=24)
        self.client = httpx.Client()

    def get_version_with_cache(self, package_name: str) -> str:
        """Get package version with cache-first strategy."""
        # Try cache first
        cached = self.cache.get(package_name)
        if cached and "version" in cached:
            print(f"Cache hit for {package_name}")
            return cached["version"]

        # Fetch from PyPI
        print(f"Cache miss for {package_name}, fetching from PyPI")
        version = get_latest_pypi_version(
            package_name,
            self.client,
            cache=self.cache
        )
        return version

    def check_packages(self, packages: list[str]):
        """Check multiple packages with cache statistics."""
        for package in packages:
            version = self.get_version_with_cache(package)
            print(f"{package}: {version}")

        # Show cache performance
        stats = self.cache.get_stats()
        print(f"\nCache Statistics:")
        print(f"  Hits: {stats['hits']}")
        print(f"  Misses: {stats['misses']}")
        print(f"  Hit rate: {stats['hits'] / (stats['hits'] + stats['misses']):.1%}")

    def close(self):
        """Clean up resources."""
        self.client.close()

# Usage
checker = CachedDependencyChecker()
try:
    checker.check_packages(["httpx", "requests", "packaging"])
finally:
    checker.close()

## Integration Examples

### Custom Script

```python
#!/usr/bin/env python3
"""Custom dependency checker using Zaojun API."""

import sys
from pathlib import Path
import httpx
from zaojun import process_dependencies, parse_dependency

def check_custom_dependencies(dependencies_file: Path) -> bool:
    """Check dependencies from a custom file format."""

    # Read custom dependencies file
    with open(dependencies_file) as f:
        dependencies = [line.strip() for line in f if line.strip()]

    # Check each dependency
    client = httpx.Client()
    try:
        needs_update = process_dependencies(
            dependencies=dependencies,
            short=True,
            compat_ok=False,
            client=client,
            cache=None,
        )
        return needs_update
    finally:
        client.close()

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: check-deps.py <dependencies-file>")
        sys.exit(1)

    deps_file = Path(sys.argv[1])
    if not deps_file.exists():
        print(f"File not found: {deps_file}")
        sys.exit(1)

    needs_update = check_custom_dependencies(deps_file)
    sys.exit(1 if needs_update else 0)

Integration with Other Tools¶

"""Integrate Zaojun with a build system."""

from pathlib import Path
import httpx
from zaojun import get_latest_pypi_version, parse_dependency

class DependencyManager:
    """Custom dependency manager using Zaojun components."""

    def __init__(self):
        self.client = httpx.Client()

    def check_and_update(self, pyproject_path: Path):
        """Check dependencies and suggest updates."""
        import tomllib

        with open(pyproject_path, "rb") as f:
            data = tomllib.load(f)

        dependencies = data.get("project", {}).get("dependencies", [])

        for dep in dependencies:
            try:
                package_name, version_spec, _ = parse_dependency(dep)
                latest = get_latest_pypi_version(package_name, self.client, cache=None)

                print(f"{package_name}: {version_spec} -> Latest: {latest}")

            except Exception as e:
                print(f"Error checking {dep}: {e}")

    def close(self):
        """Clean up resources."""
        self.client.close()

# Usage
manager = DependencyManager()
try:
    manager.check_and_update(Path("pyproject.toml"))
finally:
    manager.close()

Performance Considerations¶

HTTP Client Reuse¶

Always reuse HTTP clients when making multiple requests:

# Good: Reuse client
client = httpx.Client()
for package in packages:
    version = get_latest_pypi_version(package, client)
client.close()

# Bad: Create new client for each request
for package in packages:
    client = httpx.Client()
    version = get_latest_pypi_version(package, client)
    client.close()

Error Handling in Loops¶

Handle errors gracefully when processing multiple dependencies:

client = httpx.Client()
for dependency in dependencies:
    try:
        needs_update = check_dependency(
            dependency=dependency,
            short=True,
            compat_ok=False,
            client=client,
            cache=None,
        )
    except Exception as e:
        print(f"Skipping {dependency}: {e}")
        continue
client.close()

Testing the API¶

Unit Testing¶

"""Example unit tests for Zaojun API."""

from unittest.mock import Mock, patch
import httpx
from zaojun import check_dependency, parse_dependency
from zaojun.cache import PyPICache

def test_parse_dependency():
    """Test dependency parsing."""
    package, version, marker = parse_dependency("package==1.0.0")
    assert package == "package"
    assert version == "==1.0.0"
    assert marker is None

def test_check_dependency():
    """Test dependency checking with mock."""
    mock_client = Mock(spec=httpx.Client)
    mock_response = Mock()
    mock_response.json.return_value = {"info": {"version": "2.0.0"}}
    mock_client.get.return_value = mock_response

    with patch("zaojun.parse_dependency") as mock_parse:
        mock_parse.return_value = ("package", "==1.0.0", None)

        needs_update = check_dependency(
            dependency="package==1.0.0",
            short=False,
            compat_ok=False,
            client=mock_client,
            cache=None,
        )

        assert needs_update is True

def test_cache_integration():
    """Test dependency checking with cache."""
    mock_client = Mock(spec=httpx.Client)
    mock_cache = Mock(spec=PyPICache)
    mock_cache.enabled = True
    mock_cache.get.return_value = {"version": "1.0.0"}

    with patch("zaojun.parse_dependency") as mock_parse:
        mock_parse.return_value = ("package", "==1.0.0", None)

        needs_update = check_dependency(
            dependency="package==1.0.0",
            short=False,
            compat_ok=False,
            client=mock_client,
            cache=mock_cache,
        )

        # Verify cache was used
        mock_cache.get.assert_called_once_with("package")
        assert needs_update is False  # Version matches cache

Cache Testing¶

"""Example unit tests for PyPICache."""

import tempfile
import time
from pathlib import Path
from zaojun.cache import PyPICache

def test_cache_hit():
    """Test cache hit scenario."""
    with tempfile.TemporaryDirectory() as temp_dir:
        cache_dir = Path(temp_dir)
        cache = PyPICache(cache_dir=cache_dir)

        # Set cache entry
        cache.set("test-package", {"version": "1.0.0"})

        # Get cache entry (should hit)
        data = cache.get("test-package")
        assert data == {"version": "1.0.0"}

        stats = cache.get_stats()
        assert stats["hits"] == 1
        assert stats["misses"] == 0

def test_cache_expiration():
    """Test cache expiration."""
    with tempfile.TemporaryDirectory() as temp_dir:
        cache_dir = Path(temp_dir)
        cache = PyPICache(cache_dir=cache_dir, ttl_hours=1)

        # Manually create expired cache entry
        import json
        cache_file = cache._get_cache_path("test-package")
        cache_file.parent.mkdir(parents=True, exist_ok=True)

        expired_data = {
            "timestamp": time.time() - 7200,  # 2 hours ago
            "package_name": "test-package",
            "data": {"version": "1.0.0"},
        }

        with open(cache_file, "w") as f:
            json.dump(expired_data, f)

        # Get should return None (expired)
        data = cache.get("test-package")
        assert data is None
        assert cache.get_stats()["expired"] == 1

Cache Implementation Details¶

Cache Storage Format¶

Cache files are stored as JSON with the following structure:

{
  "timestamp": 1678886400.123456,
  "package_name": "package-name",
  "data": {
    "version": "1.0.0",
    "package_info": {
      "summary": "Package description",
      "author": "Author Name",
      "home_page": "https://example.com"
    }
  }
}

Cache Key Generation¶

Cache keys are generated using: - Normalized package names (lowercase, underscores to hyphens) - MD5 hash for filesystem safety (not cryptographic security) - JSON file extension

Thread Safety¶

The PyPICache class is designed for single-threaded use. For multi-threaded applications: - Create separate cache instances per thread - Or implement external locking mechanisms - Atomic file operations prevent corruption

Platform Compatibility¶

The cache system works on all major platforms: - Linux/Unix: Uses ~/.cache/zaojun/pypi_cache/ (XDG-compliant) - macOS: Uses ~/Library/Caches/zaojun/pypi_cache/ - Windows: Uses %LOCALAPPDATA%\zaojun\pypi_cache\

API Reference¶

Overview¶

Main Functions¶

check()¶

process_dependencies()¶

check_dependency()¶

parse_dependency()¶

is_version_compatible()¶

is_latest_version()¶

is_local_dependency()¶

get_latest_pypi_version()¶

CLI Interface¶

main()¶

app¶

Cache API¶

PyPICache Class¶

get_default_cache()¶

Types and Constants¶

Import Statements¶

Type Aliases¶

Error Handling¶

Common Exceptions¶

Error Recovery¶

Cache-Specific Error Handling¶

Cache Performance Considerations¶

Cache Configuration¶

Cache Integration Example¶

Integration with Other Tools¶

Performance Considerations¶

HTTP Client Reuse¶

Error Handling in Loops¶

Testing the API¶

Unit Testing¶

Cache Testing¶

Cache Implementation Details¶

Cache Storage Format¶

Cache Key Generation¶

Thread Safety¶

Platform Compatibility¶

See Also¶

`check()`¶

`process_dependencies()`¶

`check_dependency()`¶

`parse_dependency()`¶

`is_version_compatible()`¶

`is_latest_version()`¶

`is_local_dependency()`¶

`get_latest_pypi_version()`¶

`main()`¶

`app`¶

`PyPICache` Class¶

`get_default_cache()`¶