""" Configuration extensions for Cloudflare integration """ import os from typing import Optional from pydantic import BaseModel, Field class CloudflareSettings(BaseModel): """Cloudflare configuration settings""" api_token: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_API_TOKEN"), description="Cloudflare API token", ) account_id: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_ACCOUNT_ID"), description="Cloudflare account ID", ) worker_url: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_WORKER_URL"), description="Cloudflare Worker URL", ) # D1 Database settings d1_database_id: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_D1_DATABASE_ID"), description="D1 database ID", ) # KV Namespace settings kv_sessions_id: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_KV_SESSIONS_ID"), description="KV namespace ID for sessions", ) kv_cache_id: Optional[str] = Field( default_factory=lambda: os.getenv("CLOUDFLARE_KV_CACHE_ID"), description="KV namespace ID for cache", ) # R2 Bucket settings r2_storage_bucket: str = Field( default_factory=lambda: os.getenv( "CLOUDFLARE_R2_STORAGE_BUCKET", "openmanus-storage" ), description="R2 storage bucket name", ) r2_assets_bucket: str = Field( default_factory=lambda: os.getenv( "CLOUDFLARE_R2_ASSETS_BUCKET", "openmanus-assets" ), description="R2 assets bucket name", ) # Connection settings timeout: int = Field(default=30, description="Request timeout in seconds") def is_configured(self) -> bool: """Check if minimum Cloudflare configuration is available""" return bool(self.api_token and self.account_id) def has_worker(self) -> bool: """Check if worker URL is configured""" return bool(self.worker_url) def has_d1(self) -> bool: """Check if D1 database is configured""" return bool(self.d1_database_id) def has_kv(self) -> bool: """Check if KV namespaces are configured""" return bool(self.kv_sessions_id and self.kv_cache_id) class HuggingFaceSettings(BaseModel): """Hugging Face configuration settings""" token: Optional[str] = Field( default_factory=lambda: os.getenv("HUGGINGFACE_TOKEN"), description="Hugging Face API token", ) cache_dir: str = Field( default_factory=lambda: os.getenv( "HF_HOME", "/app/OpenManus/.cache/huggingface" ), description="Hugging Face cache directory", ) model_cache_size: int = Field( default=5, description="Maximum number of models to cache" ) class DeploymentSettings(BaseModel): """Deployment-specific settings""" environment: str = Field( default_factory=lambda: os.getenv("ENVIRONMENT", "development"), description="Deployment environment", ) debug: bool = Field( default_factory=lambda: os.getenv("DEBUG", "false").lower() == "true", description="Enable debug mode", ) log_level: str = Field( default_factory=lambda: os.getenv("LOG_LEVEL", "INFO"), description="Logging level", ) # Gradio settings server_name: str = Field( default_factory=lambda: os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"), description="Gradio server name", ) server_port: int = Field( default_factory=lambda: int(os.getenv("GRADIO_SERVER_PORT", "7860")), description="Gradio server port", ) # Security settings secret_key: Optional[str] = Field( default_factory=lambda: os.getenv("SECRET_KEY"), description="Secret key for sessions", ) jwt_secret: Optional[str] = Field( default_factory=lambda: os.getenv("JWT_SECRET"), description="JWT signing secret", ) # Create global instances cloudflare_config = CloudflareSettings() huggingface_config = HuggingFaceSettings() deployment_config = DeploymentSettings()