Yeetek's picture
Upload 17 files
b3e0a65 verified
"""
Configuration management for Anthropic Topic Segmentation Microservice.
This module handles all environment variables, model settings, and application
configuration using Pydantic Settings for type safety and validation.
"""
import os
from typing import Optional, List
from pydantic import Field, field_validator, ConfigDict
from pydantic_settings import BaseSettings
from enum import Enum
class LogLevel(str, Enum):
"""Supported logging levels."""
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
class AnthropicModel(str, Enum):
"""Supported Anthropic models."""
CLAUDE_3_5_SONNET = "claude-3-5-sonnet-20241022"
CLAUDE_3_5_HAIKU = "claude-3-5-haiku-20241022"
CLAUDE_3_SONNET = "claude-3-5-sonnet-20241022" # Updated to current version
CLAUDE_3_HAIKU = "claude-3-haiku-20240307" # Deprecated but kept for compatibility
class Settings(BaseSettings):
"""Application settings with environment variable support."""
# Application Configuration
app_name: str = Field(default="Anthropic Topic Segmentation Microservice", alias="APP_NAME")
app_version: str = Field(default="1.0.0", alias="APP_VERSION")
debug: bool = Field(default=False, alias="DEBUG")
log_level: LogLevel = Field(default=LogLevel.INFO, alias="LOG_LEVEL")
# Anthropic API Configuration
anthropic_api_key: str = Field(alias="ANTHROPIC_API_KEY", description="Anthropic API key")
anthropic_model: AnthropicModel = Field(
default=AnthropicModel.CLAUDE_3_5_SONNET,
alias="ANTHROPIC_MODEL",
description="Anthropic model to use"
)
# API Limits and Performance
max_sentences: int = Field(default=1500, ge=1, le=2000, alias="MAX_SENTENCES", description="Maximum sentences per request")
request_timeout: int = Field(default=300, ge=30, le=600, alias="REQUEST_TIMEOUT", description="Request timeout in seconds")
rate_limit_requests: int = Field(default=100, ge=1, alias="RATE_LIMIT_REQUESTS", description="Rate limit requests per window")
rate_limit_window: int = Field(default=3600, ge=60, alias="RATE_LIMIT_WINDOW", description="Rate limit window in seconds")
# Retry Configuration
max_retries: int = Field(default=3, ge=1, le=10, alias="MAX_RETRIES", description="Maximum retry attempts")
retry_delay: float = Field(default=1.0, ge=0.1, le=10.0, alias="RETRY_DELAY", description="Retry delay in seconds")
# HuggingFace Spaces Detection
space_id: Optional[str] = Field(default=None, alias="SPACE_ID", description="HuggingFace Space ID")
space_author_name: Optional[str] = Field(default=None, alias="SPACE_AUTHOR_NAME", description="HuggingFace Space author")
# CORS Configuration
cors_origins: List[str] = Field(
default=["*"],
alias="CORS_ORIGINS",
description="Comma-separated list of allowed CORS origins"
)
@field_validator("cors_origins", mode="before")
@classmethod
def parse_cors_origins(cls, v):
"""Parse CORS origins from comma-separated string or JSON array."""
if isinstance(v, str):
# Try to parse as JSON first
try:
import json
return json.loads(v)
except json.JSONDecodeError:
# If not JSON, treat as comma-separated string
return [origin.strip() for origin in v.split(",")]
return v
@field_validator("anthropic_api_key")
@classmethod
def validate_api_key(cls, v):
"""Validate Anthropic API key format."""
if not v or len(v) < 10:
raise ValueError("ANTHROPIC_API_KEY must be provided and valid")
return v
@property
def is_huggingface_spaces(self) -> bool:
"""Check if running in HuggingFace Spaces environment."""
return self.space_id is not None or os.getenv("SPACE_ID") is not None
@property
def is_development(self) -> bool:
"""Check if running in development mode."""
return self.debug or self.log_level == LogLevel.DEBUG
model_config = ConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
env_prefix="",
extra="ignore",
env_parse_none_str="None"
)
# Global settings instance - will be initialized when needed
_settings: Optional[Settings] = None
def get_settings() -> Settings:
"""Get application settings instance."""
global _settings
if _settings is None:
_settings = Settings()
return _settings
def get_anthropic_config() -> dict:
"""Get Anthropic-specific configuration."""
settings = get_settings()
return {
"api_key": settings.anthropic_api_key,
"model": settings.anthropic_model.value,
"max_retries": settings.max_retries,
"timeout": settings.request_timeout,
}
def get_api_config() -> dict:
"""Get API-specific configuration."""
settings = get_settings()
return {
"max_sentences": settings.max_sentences,
"request_timeout": settings.request_timeout,
"rate_limit_requests": settings.rate_limit_requests,
"rate_limit_window": settings.rate_limit_window,
"cors_origins": settings.cors_origins,
}