Spaces:
Sleeping
Sleeping
| """Application settings and configuration | |
| Loads environment variables and provides type-safe configuration. | |
| """ | |
| from typing import List, Union, Optional | |
| from pydantic import field_validator | |
| from pydantic_settings import BaseSettings, SettingsConfigDict | |
| from openai import AsyncOpenAI | |
| class Settings(BaseSettings): | |
| """Application settings loaded from environment variables""" | |
| model_config = SettingsConfigDict( | |
| env_file=".env", | |
| env_file_encoding="utf-8", | |
| case_sensitive=False, | |
| extra="ignore" | |
| ) | |
| # Database | |
| database_url: str | |
| # Qdrant Vector Database | |
| qdrant_url: str | |
| qdrant_api_key: str | |
| qdrant_collection_name: str = "humanoid-robotics-book-v1" | |
| vector_size: int = 1536 # OpenAI text-embedding-3-small dimension | |
| # OpenAI | |
| openai_api_key: str | |
| openai_embedding_model: str = "text-embedding-3-small" | |
| chat_model: str = "gpt-4o-mini" # Fast, cost-effective model for chat (was gpt-4-turbo-preview) | |
| # Authentication | |
| better_auth_secret: str | |
| session_expiry_days: int = 7 | |
| # Rate Limiting | |
| rate_limit_per_minute: int = 20 | |
| redis_url: str = "redis://localhost:6379" | |
| # CORS | |
| allowed_origins: Union[str, List[str]] = "http://localhost:3000,http://localhost:8000" | |
| # Application | |
| environment: str = "development" | |
| log_level: str = "INFO" | |
| def parse_cors_origins(cls, v): | |
| """Parse CORS origins from comma-separated string or list""" | |
| if isinstance(v, str): | |
| return [origin.strip() for origin in v.split(",")] | |
| return v | |
| def is_production(self) -> bool: | |
| """Check if running in production environment""" | |
| return self.environment.lower() == "production" | |
| def async_database_url(self) -> str: | |
| """Get async database URL for SQLAlchemy | |
| Converts postgresql:// to postgresql+asyncpg:// and removes sslmode and | |
| channel_binding parameters since asyncpg uses different SSL configuration. | |
| """ | |
| url = self.database_url | |
| # Replace postgresql:// with postgresql+asyncpg:// | |
| if url.startswith("postgresql://"): | |
| url = url.replace("postgresql://", "postgresql+asyncpg://", 1) | |
| # Remove sslmode and channel_binding parameters that asyncpg doesn't support | |
| # asyncpg will handle SSL automatically | |
| if "sslmode=" in url or "channel_binding=" in url: | |
| from urllib.parse import urlparse, parse_qs, urlencode, urlunparse | |
| parsed = urlparse(url) | |
| query_params = parse_qs(parsed.query) | |
| # Remove sslmode and channel_binding | |
| query_params.pop('sslmode', None) | |
| query_params.pop('channel_binding', None) | |
| # Reconstruct the query string | |
| new_query = urlencode(query_params, doseq=True) | |
| url = urlunparse(( | |
| parsed.scheme, | |
| parsed.netloc, | |
| parsed.path, | |
| parsed.params, | |
| new_query, | |
| parsed.fragment | |
| )) | |
| return url | |
| # Global settings instance | |
| settings = Settings() | |
| # Global OpenAI client instance (only if API key is provided) | |
| openai_client = AsyncOpenAI(api_key=settings.openai_api_key) if settings.openai_api_key else None | |