47a710a8b9
- Update all datetime.utcnow() to datetime.now(UTC) for Python 3.12+ compatibility - Update all datetime.utcfromtimestamp() to datetime.fromtimestamp(..., UTC) - Fix dataclass default_factory to use lambda: datetime.now(UTC) - Update auth, audit, billing, and realtime modules - Add UTC import from datetime module - Update pytest configuration to include pytest-asyncio - All 28 Phase 8 enterprise tests pass with no warnings Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
"""비용 관리 모델.
|
|
|
|
Phase 8: 사용량 및 구독 관리
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, UTC
|
|
from enum import Enum
|
|
from typing import Optional, Dict, Any
|
|
import uuid
|
|
|
|
|
|
class OperationType(str, Enum):
|
|
"""작업 타입."""
|
|
|
|
LLM_CALL = "llm_call" # LLM 호출 (토큰 기반)
|
|
LLM_STREAM = "llm_stream" # 스트리밍 (분 기반)
|
|
GRAPH_QUERY = "graph_query" # 그래프 쿼리 (노드 기반)
|
|
STORAGE = "storage" # 저장소 (GB 기반)
|
|
API_CALL = "api_call" # API 호출 (호출 수)
|
|
ANALYSIS = "analysis" # 분석 (작업)
|
|
|
|
|
|
class SubscriptionTier(str, Enum):
|
|
"""구독 계층."""
|
|
|
|
FREE = "free"
|
|
PRO = "pro"
|
|
ENTERPRISE = "enterprise"
|
|
|
|
|
|
@dataclass
|
|
class Usage:
|
|
"""사용량 기록."""
|
|
|
|
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
|
org_id: str = ""
|
|
user_id: str = ""
|
|
operation_type: OperationType = OperationType.API_CALL
|
|
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
quantity: float = 0.0 # 토큰, 노드, GB, 시간 등
|
|
cost: float = 0.0 # USD
|
|
metadata: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""딕셔너리로 변환."""
|
|
return {
|
|
"id": self.id,
|
|
"org_id": self.org_id,
|
|
"user_id": self.user_id,
|
|
"operation_type": self.operation_type.value,
|
|
"timestamp": self.timestamp.isoformat(),
|
|
"quantity": self.quantity,
|
|
"cost": self.cost,
|
|
"metadata": self.metadata,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Subscription:
|
|
"""구독 정보."""
|
|
|
|
org_id: str = ""
|
|
tier: SubscriptionTier = SubscriptionTier.FREE
|
|
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
auto_renew: bool = True
|
|
current_month_cost: float = 0.0
|
|
monthly_limit: float = 100.0 # USD
|
|
exceeded_limit: bool = False
|
|
metadata: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""딕셔너리로 변환."""
|
|
return {
|
|
"org_id": self.org_id,
|
|
"tier": self.tier.value,
|
|
"created_at": self.created_at.isoformat(),
|
|
"current_month_cost": self.current_month_cost,
|
|
"monthly_limit": self.monthly_limit,
|
|
"exceeded_limit": self.exceeded_limit,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class UsageStatistics:
|
|
"""사용량 통계."""
|
|
|
|
period_start: datetime
|
|
period_end: datetime
|
|
total_cost: float = 0.0
|
|
by_operation_type: Dict[str, float] = field(default_factory=dict)
|
|
by_user: Dict[str, float] = field(default_factory=dict)
|
|
api_calls: int = 0
|
|
llm_tokens: float = 0
|
|
storage_gb: float = 0.0
|
|
|
|
def to_dict(self) -> dict:
|
|
"""딕셔너리로 변환."""
|
|
return {
|
|
"period_start": self.period_start.isoformat(),
|
|
"period_end": self.period_end.isoformat(),
|
|
"total_cost": self.total_cost,
|
|
"by_operation_type": self.by_operation_type,
|
|
"by_user": self.by_user,
|
|
"api_calls": self.api_calls,
|
|
"llm_tokens": self.llm_tokens,
|
|
"storage_gb": self.storage_gb,
|
|
}
|