Skip to content

ccproxy.plugins.pricing.service

ccproxy.plugins.pricing.service

Pricing service providing unified interface for pricing functionality.

PricingService 

PricingService(config)

Main service interface for pricing functionality.

Source code in ccproxy/plugins/pricing/service.py 
26
27
28
29
30
31
32
def __init__(self, config: PricingConfig):
    """Initialize pricing service with configuration."""
    self.config = config
    self.cache = PricingCache(config)
    self.loader = PricingLoader()
    self.updater = PricingUpdater(self.cache, config)
    self._current_pricing: PricingData | None = None

initialize async 

initialize()

Initialize the pricing service.

Source code in ccproxy/plugins/pricing/service.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
async def initialize(self) -> None:
    """Initialize the pricing service."""
    if not self.config.enabled:
        logger.info("pricing_service_disabled")
        return

    logger.debug("pricing_service_initializing")

    # Force refresh on startup if configured
    if self.config.force_refresh_on_startup:
        await self.force_refresh_pricing()
    else:
        # Load current pricing data
        await self.get_current_pricing()

get_current_pricing async 

get_current_pricing(force_refresh=False)

Get current pricing data.

Source code in ccproxy/plugins/pricing/service.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
async def get_current_pricing(
    self, force_refresh: bool = False
) -> PricingData | None:
    """Get current pricing data."""
    if not self.config.enabled:
        return None

    if force_refresh or self._current_pricing is None:
        self._current_pricing = await self.updater.get_current_pricing(
            force_refresh
        )

    return self._current_pricing

get_model_pricing async 

get_model_pricing(model_name)

Get pricing for specific model.

Source code in ccproxy/plugins/pricing/service.py 
63
64
65
66
67
68
69
async def get_model_pricing(self, model_name: str) -> ModelPricing | None:
    """Get pricing for specific model."""
    pricing_data = await self.get_current_pricing()
    if pricing_data is None:
        return None

    return pricing_data.get(model_name)

calculate_cost async 

calculate_cost(
    model_name,
    input_tokens=0,
    output_tokens=0,
    cache_read_tokens=0,
    cache_write_tokens=0,
)

Calculate cost for token usage.

Raises:

Type Description
PricingServiceDisabledError

If pricing service is disabled
ModelPricingNotFoundError

If model pricing is not found
Source code in ccproxy/plugins/pricing/service.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def calculate_cost(
    self,
    model_name: str,
    input_tokens: int = 0,
    output_tokens: int = 0,
    cache_read_tokens: int = 0,
    cache_write_tokens: int = 0,
) -> Decimal:
    """Calculate cost for token usage.

    Raises:
        PricingServiceDisabledError: If pricing service is disabled
        ModelPricingNotFoundError: If model pricing is not found
    """
    if not self.config.enabled:
        raise PricingServiceDisabledError()

    model_pricing = await self.get_model_pricing(model_name)
    if model_pricing is None:
        raise ModelPricingNotFoundError(model_name)

    # Calculate cost per million tokens, then scale to actual tokens
    total_cost = Decimal("0")

    if input_tokens > 0:
        total_cost += (model_pricing.input * input_tokens) / Decimal("1000000")

    if output_tokens > 0:
        total_cost += (model_pricing.output * output_tokens) / Decimal("1000000")

    if cache_read_tokens > 0:
        total_cost += (model_pricing.cache_read * cache_read_tokens) / Decimal(
            "1000000"
        )

    if cache_write_tokens > 0:
        total_cost += (model_pricing.cache_write * cache_write_tokens) / Decimal(
            "1000000"
        )

    return total_cost

calculate_cost_sync 

calculate_cost_sync(
    model_name,
    input_tokens=0,
    output_tokens=0,
    cache_read_tokens=0,
    cache_write_tokens=0,
)

Calculate cost synchronously using cached pricing data.

This method uses the cached pricing data and doesn't make any async calls, making it safe to use in streaming contexts where we can't await.

Raises:

Type Description
PricingServiceDisabledError

If pricing service is disabled
PricingDataNotLoadedError

If pricing data is not loaded yet
ModelPricingNotFoundError

If model pricing is not found
Source code in ccproxy/plugins/pricing/service.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def calculate_cost_sync(
    self,
    model_name: str,
    input_tokens: int = 0,
    output_tokens: int = 0,
    cache_read_tokens: int = 0,
    cache_write_tokens: int = 0,
) -> Decimal:
    """Calculate cost synchronously using cached pricing data.

    This method uses the cached pricing data and doesn't make any async calls,
    making it safe to use in streaming contexts where we can't await.

    Raises:
        PricingServiceDisabledError: If pricing service is disabled
        PricingDataNotLoadedError: If pricing data is not loaded yet
        ModelPricingNotFoundError: If model pricing is not found
    """
    if not self.config.enabled:
        raise PricingServiceDisabledError()

    if self._current_pricing is None:
        raise PricingDataNotLoadedError()

    model_pricing = self._current_pricing.get(model_name)
    if model_pricing is None:
        raise ModelPricingNotFoundError(model_name)

    # Calculate cost per million tokens, then scale to actual tokens
    total_cost = Decimal("0")

    if input_tokens > 0:
        total_cost += (model_pricing.input * input_tokens) / Decimal("1000000")

    if output_tokens > 0:
        total_cost += (model_pricing.output * output_tokens) / Decimal("1000000")

    if cache_read_tokens > 0:
        total_cost += (model_pricing.cache_read * cache_read_tokens) / Decimal(
            "1000000"
        )

    if cache_write_tokens > 0:
        total_cost += (model_pricing.cache_write * cache_write_tokens) / Decimal(
            "1000000"
        )

    return total_cost

force_refresh_pricing async 

force_refresh_pricing()

Force refresh of pricing data.

Source code in ccproxy/plugins/pricing/service.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
async def force_refresh_pricing(self) -> bool:
    """Force refresh of pricing data."""
    if not self.config.enabled:
        return False

    success = await self.updater.force_refresh()
    if success:
        # Reload the current pricing data after successful refresh
        self._current_pricing = await self.updater.get_current_pricing(
            force_refresh=True
        )
        return True
    return False

get_available_models async 

get_available_models()

Get list of available models with pricing.

Source code in ccproxy/plugins/pricing/service.py 
176
177
178
179
180
181
182
async def get_available_models(self) -> list[str]:
    """Get list of available models with pricing."""
    pricing_data = await self.get_current_pricing()
    if pricing_data is None:
        return []

    return pricing_data.model_names()

get_cache_info 

get_cache_info()

Get cache status information.

Source code in ccproxy/plugins/pricing/service.py 
184
185
186
def get_cache_info(self) -> dict[str, Any]:
    """Get cache status information."""
    return self.cache.get_cache_info()

clear_cache async 

clear_cache()

Clear pricing cache.

Source code in ccproxy/plugins/pricing/service.py 
188
189
190
191
async def clear_cache(self) -> bool:
    """Clear pricing cache."""
    self._current_pricing = None
    return self.cache.clear_cache()