mindroot.coreplugins.credits package

Submodules

mindroot.coreplugins.credits.conversion module

class mindroot.coreplugins.credits.conversion.CreditPolicy(ledger: CreditLedger, ratio_config: CreditRatioConfig, base_path: str)

Bases: object

Encapsulates credit-related business rules and policies

async check_operation_allowed(username: str, plugin_id: str, cost_type_id: str, estimated_cost: float, model_id: str | None = None) → bool

Check if an operation should be allowed based on estimated cost

async estimate_credits_needed(plugin_id: str, cost_type_id: str, estimated_cost: float, model_id: str | None = None) → float

Estimate credits needed for an operation

async get_user_limits(username: str) → Dict

Get user’s credit limits and current usage

class mindroot.coreplugins.credits.conversion.CreditUsageHandler(ledger: CreditLedger, ratio_config: CreditRatioConfig, base_path: str)

Bases: object

Converts usage costs to credits and records credit transactions.

async handle_usage(plugin_id: str, cost_type_id: str, quantity: float, metadata: dict, context=None, model_id: str | None = None) → None

Handle a usage event by converting cost to credits and recording transaction

mindroot.coreplugins.credits.ledger module

class mindroot.coreplugins.credits.ledger.CreditLedger(storage: CreditStorage)

Bases: object

Manages credit transactions and balances. Ensures consistency and provides high-level credit operations.

async check_credits_available(username: str, required_amount: float) → Tuple[bool, float]

Check if user has sufficient credits

Returns:

has_sufficient, float: current_balance)

Return type:

Tuple of (bool

async get_balance(username: str) → float

Get current credit balance

async get_balance_at(username: str, at_date: date) → float

Get credit balance at a specific date

async get_transactions(username: str, start_date: date | None = None, end_date: date | None = None) → List[CreditTransaction]

Get transaction history

async get_usage_summary(username: str, start_date: date | None = None, end_date: date | None = None) → Dict

Get summary of credit usage

async record_allocation(username: str, amount: float, source: str, reference_id: str, metadata: Dict | None = None) → float

Record a credit allocation (positive credit change)

async record_refund(username: str, amount: float, source: str, reference_id: str, original_transaction_id: str, metadata: Dict | None = None) → float

Record a credit refund (reversal of usage)

async record_usage(username: str, amount: float, source: str, reference_id: str, metadata: Dict | None = None, allow_negative: bool = False) → float

Record a credit usage (negative credit change)

exception mindroot.coreplugins.credits.ledger.InsufficientCreditsError

Bases: Exception

mindroot.coreplugins.credits.mod module

class mindroot.coreplugins.credits.mod.CreditsPlugin(base_path: str)

Bases: object

create_components()

Create fresh instances of all credit system components

async mindroot.coreplugins.credits.mod.allocate_credits(username: str, amount: float, source: str, reference_id: str, metadata: Dict | None = None, context=None) → float

Allocate credits to a user.

Parameters:

• username – User to allocate credits to

• amount – Amount of credits to allocate

• source – Source of allocation (e.g., ‘purchase’, ‘admin_grant’)

• reference_id – External reference (e.g., payment ID)

• metadata – Additional information about allocation

• context – Request context

Returns:

New credit balance

Return type:

float

Example

new_balance = await allocate_credits(

‘user123’, 1000.0, ‘purchase’, ‘payment_123’, {‘payment_method’: ‘stripe’}

)

async mindroot.coreplugins.credits.mod.check_credits_available(username: str, required_amount: float, context=None) → Dict

Check if user has sufficient credits.

Returns:

Dict with ‘has_sufficient’ and ‘current_balance’ keys

async mindroot.coreplugins.credits.mod.estimate_credits(plugin_id: str, cost_type_id: str, estimated_cost: float, model_id: str | None = None, context=None) → Dict

Estimate credits needed for an operation.

Example

estimate = await estimate_credits(

‘gpt4’, ‘gpt4.input_tokens’, 0.01, # $0.01 ‘gpt-4-1106-preview’

)

mindroot.coreplugins.credits.mod.get_base_path(context) → str

Get the base path for credit data storage

async mindroot.coreplugins.credits.mod.get_credit_balance(username: str, context=None) → float

Get current credit balance for a user.

Example

balance = await get_credit_balance(‘user123’)

async mindroot.coreplugins.credits.mod.get_credit_ratios(context=None) → Dict

Get current credit ratio configuration.

async mindroot.coreplugins.credits.mod.get_credit_report(username: str, start_date: str | None = None, end_date: str | None = None, context=None) → Dict

Get detailed credit report for a user.

Parameters:

• username – User to get report for

• start_date – Optional start date (YYYY-MM-DD)

• end_date – Optional end date (YYYY-MM-DD)

Returns:

• Current balance

• Transaction history

• Usage summary

Return type:

Dict containing

async mindroot.coreplugins.credits.mod.handle_usage(plugin_id: str, cost_type_id: str, quantity: float, metadata: dict, context=None, model_id: str | None = None)

Handle usage tracking for credits system. This service is called by the usage plugin after tracking usage.

async mindroot.coreplugins.credits.mod.set_credit_ratio(ratio: float, plugin_id: str | None = None, cost_type_id: str | None = None, model_id: str | None = None, context=None)

Set credit ratio for cost conversion.

Parameters:

• ratio – Credits per unit cost

• plugin_id – Optional plugin identifier

• cost_type_id – Optional cost type identifier

• model_id – Optional model identifier

Example

# Set global default ratio await set_credit_ratio(100.0)

# Set model-specific ratio await set_credit_ratio(

90.0, plugin_id=’gpt4’, cost_type_id=’gpt4.input_tokens’, model_id=’gpt-4-1106-preview’

)

async mindroot.coreplugins.credits.mod.startup(app, context=None)

Startup tasks

mindroot.coreplugins.credits.models module

class mindroot.coreplugins.credits.models.CreditRatioConfig(base_path: str)

Bases: object

Manages the hierarchical credit ratio configuration

async get_config() → Dict

Get the full configuration

async get_ratio(plugin_id: str, cost_type_id: str, model_id: str | None = None) → float

Get the appropriate credit ratio following the resolution order

async set_config(config: Dict) → None

Set the full configuration

async set_ratio(ratio: float, plugin_id: str | None = None, cost_type_id: str | None = None, model_id: str | None = None) → None

Set a credit ratio at the specified level

class mindroot.coreplugins.credits.models.CreditTransaction(timestamp: datetime.datetime, username: str, amount: float, balance: float, type: str, source: str, reference_id: str, metadata: Dict[str, Any] = <factory>, transaction_id: str = <factory>)

Bases: object

amount: float

balance: float

classmethod from_dict(data: dict) → CreditTransaction

metadata: Dict[str, Any]

reference_id: str

source: str

timestamp: datetime

to_dict() → dict

transaction_id: str

type: str

username: str

mindroot.coreplugins.credits.router module

async mindroot.coreplugins.credits.router.api_allocate_credits(request: Request)

Allocate credits to a user

Request body: {

“username”: str, “amount”: float, “source”: str, “reference_id”: str, “metadata”: dict (optional)

}

async mindroot.coreplugins.credits.router.api_credit_report(username: str, start_date: str | None = None, end_date: str | None = None, request: Request = None)

Get credit report for a user

Path parameters:

username: str - Username to get report for

Query parameters:

start_date: Optional[str] - Start date in YYYY-MM-DD format end_date: Optional[str] - End date in YYYY-MM-DD format

async mindroot.coreplugins.credits.router.api_estimate_credits(plugin_id: str, cost_type_id: str, estimated_cost: float, model_id: str | None = None, request: Request = None)

Estimate credits needed for an operation

Query parameters:

plugin_id: str - Plugin identifier cost_type_id: str - Cost type identifier estimated_cost: float - Estimated cost in base currency model_id: Optional[str] - Model identifier

async mindroot.coreplugins.credits.router.api_update_ratio(request: Request)

Update credit ratio configuration

Request body: {

“ratio”: float, “plugin_id”: str (optional), “cost_type_id”: str (optional), “model_id”: str (optional)

}

async mindroot.coreplugins.credits.router.credits_admin(request: Request)

Admin interface for credit management

async mindroot.coreplugins.credits.router.credits_ratio_admin(request: Request)

Admin interface for credit ratio configuration

mindroot.coreplugins.credits.storage module

class mindroot.coreplugins.credits.storage.CreditStorage(base_path: str)

Bases: object

async get_balance_at(username: str, at_date: date) → float

Get the user’s balance at a specific date

async get_latest_balance(username: str) → float

Get the user’s latest balance

async get_transactions(username: str, start_date: date | None = None, end_date: date | None = None) → List[CreditTransaction]

Get credit transactions for a user within date range

async store_transaction(transaction: CreditTransaction) → None

Store a credit transaction

Module contents