ccproxy.utils.version_checker

ccproxy.utils.version_checker

Version checking utilities for ccproxy.

VersionCheckState

Bases: BaseModel

State tracking for version checks.

fetch_latest_github_version async

fetch_latest_github_version()

Fetch the latest version from GitHub releases API.

Returns:

Type	Description
str | None	Latest version string or None if unable to fetch

Source code in ccproxy/utils/version_checker.py

async def fetch_latest_github_version() -> str | None:
    """
    Fetch the latest version from GitHub releases API.

    Returns:
        Latest version string or None if unable to fetch
    """
    url = "https://api.github.com/repos/CaddyGlow/ccproxy-api/releases/latest"
    headers = {
        "User-Agent": f"ccproxy-api/{__version__}",
        "Accept": "application/vnd.github.v3+json",
    }

    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.get(url, headers=headers)
            response.raise_for_status()

            data: dict[str, Any] = response.json()
            tag_name: str = str(data.get("tag_name", "")).lstrip("v")

            if tag_name:
                logger.debug("github_version_fetched", latest_version=tag_name)
                return tag_name

            logger.warning("github_version_missing_tag")
            return None

    except httpx.TimeoutException:
        logger.warning("github_version_timeout")
        return None
    except httpx.HTTPStatusError as e:
        logger.warning("github_version_http_error", status_code=e.response.status_code)
        return None
    except Exception as e:
        logger.warning(
            "github_version_fetch_failed",
            error=str(e),
            error_type=type(e).__name__,
        )
        return None

get_current_version

get_current_version()

Get the current version of ccproxy.

Returns:

Type	Description
str	Current version string

Source code in ccproxy/utils/version_checker.py

def get_current_version() -> str:
    """
    Get the current version of ccproxy.

    Returns:
        Current version string
    """
    return __version__

compare_versions

compare_versions(current, latest)

Compare version strings to determine if an update is available.

Parameters:

Name	Type	Description	Default
current	str	Current version string	required
latest	str	Latest version string	required

Returns:

Type	Description
bool	True if latest version is newer than current

Source code in ccproxy/utils/version_checker.py

def compare_versions(current: str, latest: str) -> bool:
    """
    Compare version strings to determine if an update is available.

    Args:
        current: Current version string
        latest: Latest version string

    Returns:
        True if latest version is newer than current
    """
    try:
        current_parsed = pkg_version.parse(current)
        latest_parsed = pkg_version.parse(latest)

        # For dev versions, compare base version instead
        if current_parsed.is_devrelease:
            current_base = pkg_version.parse(current_parsed.base_version)
            return latest_parsed > current_base

        return latest_parsed > current_parsed
    except Exception as e:
        logger.error(
            "version_comparison_failed",
            current=current,
            latest=latest,
            error=str(e),
            error_type=type(e).__name__,
        )
        return False

load_check_state async

load_check_state(path)

Load version check state from file.

Parameters:

Name	Type	Description	Default
path	Path	Path to state file	required

Returns:

Type	Description
VersionCheckState | None	VersionCheckState if file exists and is valid, None otherwise

Source code in ccproxy/utils/version_checker.py

async def load_check_state(path: Path) -> VersionCheckState | None:
    """
    Load version check state from file.

    Args:
        path: Path to state file

    Returns:
        VersionCheckState if file exists and is valid, None otherwise
    """
    if not path.exists():
        return None

    try:
        async with aiofiles.open(path) as f:
            content = await f.read()
            data = json.loads(content)
            return VersionCheckState(**data)
    except Exception as e:
        logger.warning(
            "version_check_state_load_failed",
            path=str(path),
            error=str(e),
            error_type=type(e).__name__,
        )
        return None

save_check_state async

save_check_state(path, state)

Save version check state to file.

Parameters:

Name	Type	Description	Default
path	Path	Path to state file	required
state	VersionCheckState	VersionCheckState to save	required

Source code in ccproxy/utils/version_checker.py

async def save_check_state(path: Path, state: VersionCheckState) -> None:
    """
    Save version check state to file.

    Args:
        path: Path to state file
        state: VersionCheckState to save
    """
    try:
        # Ensure directory exists
        path.parent.mkdir(parents=True, exist_ok=True)

        # Convert state to dict with ISO format datetime
        state_dict = state.model_dump()
        state_dict["last_check_at"] = state.last_check_at.isoformat()

        async with aiofiles.open(path, "w") as f:
            await f.write(json.dumps(state_dict, indent=2))

        logger.debug("version_check_state_saved", path=str(path))
    except Exception as e:
        logger.warning(
            "version_check_state_save_failed",
            path=str(path),
            error=str(e),
            error_type=type(e).__name__,
        )

get_version_check_state_path

get_version_check_state_path()

Get the path to the version check state file.

Returns:

Type	Description
Path	Path to version_check.json in ccproxy config directory

Source code in ccproxy/utils/version_checker.py

def get_version_check_state_path() -> Path:
    """
    Get the path to the version check state file.

    Returns:
        Path to version_check.json in ccproxy config directory
    """
    return get_ccproxy_config_dir() / "version_check.json"