ccproxy.pricing.updater

ccproxy.pricing.updater

Pricing updater for managing periodic refresh of pricing data.

PricingUpdater

PricingUpdater(cache, settings)

Manages periodic updates of pricing data.

Parameters:

Name	Type	Description	Default
cache	PricingCache	Pricing cache instance	required
settings	PricingSettings	Pricing configuration settings	required

Source code in ccproxy/pricing/updater.py

def __init__(
    self,
    cache: PricingCache,
    settings: PricingSettings,
) -> None:
    """Initialize pricing updater.

    Args:
        cache: Pricing cache instance
        settings: Pricing configuration settings
    """
    self.cache = cache
    self.settings = settings
    self._cached_pricing: PricingData | None = None
    self._last_load_time: float = 0
    self._last_file_check_time: float = 0
    self._cached_file_mtime: float = 0

get_current_pricing async

get_current_pricing(force_refresh=False)

Get current pricing data with automatic updates.

Parameters:

Name	Type	Description	Default
force_refresh	bool	Force refresh even if cache is valid	False

Returns:

Type	Description
PricingData | None	Current pricing data as PricingData model

Source code in ccproxy/pricing/updater.py

async def get_current_pricing(
    self, force_refresh: bool = False
) -> PricingData | None:
    """Get current pricing data with automatic updates.

    Args:
        force_refresh: Force refresh even if cache is valid

    Returns:
        Current pricing data as PricingData model
    """
    import time

    current_time = time.time()

    # Return cached pricing if recent and not forced
    if (
        not force_refresh
        and self._cached_pricing is not None
        and (current_time - self._last_load_time) < self.settings.memory_cache_ttl
    ):
        # Only check file changes every 30 seconds to reduce I/O
        if (current_time - self._last_file_check_time) > 30:
            if self._has_cache_file_changed():
                logger.info("cache_file_changed")
                # File changed, need to reload
                pricing_data = await self._load_pricing_data()
                self._cached_pricing = pricing_data
                self._last_load_time = current_time
                return pricing_data
            self._last_file_check_time = current_time

        return self._cached_pricing

    # Check if we need to refresh
    should_refresh = force_refresh or (
        self.settings.auto_update and not self.cache.is_cache_valid()
    )

    if should_refresh:
        logger.info("pricing_refresh_start")
        await self._refresh_pricing()

    # Load pricing data
    pricing_data = await self._load_pricing_data()

    # Cache the result
    self._cached_pricing = pricing_data
    self._last_load_time = current_time
    self._last_file_check_time = current_time

    return pricing_data

force_refresh async

force_refresh()

Force a refresh of pricing data.

Returns:

Type	Description
bool	True if refresh was successful

Source code in ccproxy/pricing/updater.py

async def force_refresh(self) -> bool:
    """Force a refresh of pricing data.

    Returns:
        True if refresh was successful
    """
    logger.info("pricing_force_refresh_start")

    # Clear cached pricing
    self._cached_pricing = None
    self._last_load_time = 0

    # Refresh from external source
    success = await self._refresh_pricing()

    if success:
        # Reload pricing data
        await self.get_current_pricing(force_refresh=True)

    return success

clear_cache

clear_cache()

Clear all cached pricing data.

Returns:

Type	Description
bool	True if cache was cleared successfully

Source code in ccproxy/pricing/updater.py

def clear_cache(self) -> bool:
    """Clear all cached pricing data.

    Returns:
        True if cache was cleared successfully
    """
    logger.info("pricing_cache_clear_start")

    # Clear in-memory cache
    self._cached_pricing = None
    self._last_load_time = 0

    # Clear file cache
    return self.cache.clear_cache()

get_pricing_info async

get_pricing_info()

Get information about current pricing state.

Returns:

Type	Description
dict[str, Any]	Dictionary with pricing information

Source code in ccproxy/pricing/updater.py

async def get_pricing_info(self) -> dict[str, Any]:
    """Get information about current pricing state.

    Returns:
        Dictionary with pricing information
    """
    cache_info = self.cache.get_cache_info()

    pricing_data = await self.get_current_pricing()

    return {
        "models_loaded": len(pricing_data) if pricing_data else 0,
        "model_names": pricing_data.model_names() if pricing_data else [],
        "auto_update": self.settings.auto_update,
        "fallback_to_embedded": self.settings.fallback_to_embedded,
        "has_cached_pricing": self._cached_pricing is not None,
    }

validate_external_source async

validate_external_source()

Validate that external pricing source is accessible.

Returns:

Type	Description
bool	True if external source is accessible and has valid data

Source code in ccproxy/pricing/updater.py

async def validate_external_source(self) -> bool:
    """Validate that external pricing source is accessible.

    Returns:
        True if external source is accessible and has valid data
    """
    try:
        logger.debug("external_pricing_validation_start")

        # Try to download data
        raw_data = await self.cache.download_pricing_data(timeout=10)
        if raw_data is None:
            return False

        # Try to parse Claude models
        claude_models = PricingLoader.extract_claude_models(raw_data)
        if not claude_models:
            logger.warning("claude_models_not_found_in_external")
            return False

        # Try to load and validate using Pydantic
        pricing_data = PricingLoader.load_pricing_from_data(raw_data, verbose=False)
        if not pricing_data:
            logger.warning("external_pricing_load_failed")
            return False

        logger.info(
            "external_pricing_validation_completed", model_count=len(pricing_data)
        )
        return True

    except Exception as e:
        logger.error("external_pricing_validation_failed", error=str(e))
        return False