928c2f20aa
- Fix text selection menu not showing: wire contentRef via inputRef on multiline TextField - Fix blog title not truncating: add min-w-0 for flex item overflow - Fix outline generation 500: escape curly braces in f-string prompt template - Fix content generation 'NoneType not callable': replace SessionLocal() with get_session_for_user(), add db param to MediumBlogGenerator, fix signature mismatch in database_task_manager - Fix writing assistant suggest 500: add auth + user_id to API endpoint and service, replace sync requests with httpx.AsyncClient - Fix hallucination detector 404: explicitly include router in main.py and app.py - Fix missing error_data in task failure responses - Hide CopilotKit web inspector button - Remove hardcoded fallback suggestions from SmartTypingAssist - Fix stale closure refs in SmartTypingAssist handleTypingChange - Add two-column editor layout, stats bar, section hover menu - Various subscription, billing, and research module improvements
39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
"""
|
|
Usage limit enforcement functions.
|
|
Extracted from usage_tracking_service.py for better maintainability.
|
|
"""
|
|
|
|
from typing import Tuple, Dict, Any
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.orm import Session
|
|
from loguru import logger
|
|
|
|
from models.subscription_models import APIProvider
|
|
from services.subscription.pricing_service import PricingService
|
|
|
|
|
|
def enforce_usage_limits(user_id: str, provider: APIProvider,
|
|
tokens_requested: int, db: Session,
|
|
pricing_service: PricingService) -> Tuple[bool, str, Dict[str, Any]]:
|
|
"""Enforce usage limits before making an API call."""
|
|
# Check short-lived cache first (30s)
|
|
cache_key = f"{user_id}:{provider.value}"
|
|
now = datetime.utcnow()
|
|
|
|
# This would need access to self._enforce_cache
|
|
# For now, keeping the structure
|
|
|
|
result = pricing_service.check_usage_limits(
|
|
user_id=user_id,
|
|
provider=provider,
|
|
tokens_requested=tokens_requested
|
|
)
|
|
|
|
# Cache the result
|
|
# self._enforce_cache[cache_key] = {
|
|
# 'result': result,
|
|
# 'expires_at': now + timedelta(seconds=30)
|
|
# }
|
|
|
|
return tuple(result)
|