repo_description: CodeGate: Security, Workspaces and Multiplexing for AI Agentic Frameworks
repo_topics: ai ai-gateway aisecurity artificial-intelligence claude cline codegen copilot developer-productivity developer-tools generative-ai ide jetbrains llm ollama openai privacy python security vscode
code: """Configuration management for codegate."""
  
  import os
  from dataclasses import dataclass, field
  from pathlib import Path
  from typing import Dict, Optional, Union
  
  import structlog
  import yaml
  
  from codegate.codegate_logging import LogFormat, LogLevel
  from codegate.exceptions import ConfigurationError
  from codegate.prompts import PromptConfig
  
  logger = structlog.get_logger("codegate")
  
  # Default provider URLs
  DEFAULT_PROVIDER_URLS = {
      "openai": "https://api.openai.com",
      "openrouter": "https://openrouter.ai/api",
      "anthropic": "https://api.anthropic.com",
      "vllm": "http://localhost:8000",  # Base URL without /v1 path
      "ollama": "http://localhost:11434",  # Default Ollama server URL
      "lm_studio": "http://localhost:1234",
      "llamacpp": "./codegate_volume/models",  # Default LlamaCpp model path
  }
  
  API_DEFAULT_PAGE_SIZE = 50
  API_MAX_PAGE_SIZE = 100
  
  
  @dataclass
  class Config:
      """Application configuration with priority resolution."""
  
      # Singleton instance of Config which is set in Config.load().
      # All consumers can call: Config.get_config() to get the config.
      __config = None
  
      port: int = 8989
      proxy_port: int = 8990  # New proxy port configuration
      host: str = "localhost"
      log_level: LogLevel = LogLevel.INFO
      log_format: LogFormat = LogFormat.JSON
      prompts: PromptConfig = field(default_factory=PromptConfig)
  
      model_base_path: str = "./codegate_volume/models"
      chat_model_n_ctx: int = 32768
      chat_model_n_gpu_layers: int = -1
      embedding_model: str = "all-minilm-L6-v2-q5_k_m.gguf"
      db_path: Optional[str] = None
      vec_db_path: Optional[str] = "./sqlite_data/vectordb.db"  # Vector database
  
      # Certificate configuration
      certs_dir: str = "./codegate_volume/certs"
      ca_cert: str = "ca.crt"
      ca_key: str = "ca.key"
      server_cert: str = "server.crt"
      server_key: str = "server.key"
      force_certs: bool = False
  
      # Update configuration.
      use_update_service: bool = False
      update_service_url: str = "https://updates.codegate.ai/api/v1/version"
  
      max_fim_hash_lifetime: int = 60 * 5  # Time in seconds. Default is 5 minutes.
  
      # Min value is 0 (max similarity), max value is 2 (orthogonal)
      # The value 0.75 was found through experimentation. See /tests/muxing/test_semantic_router.py
      # It's the threshold value to determine if a query matches a persona.
      persona_threshold = 0.75
      # The value 0.3 was found through experimentation. See /tests/muxing/test_semantic_router.py
      # It's the threshold value to determine if a persona description is similar to existing personas
      persona_diff_desc_threshold = 0.3
  
      # Provider URLs with defaults
      provider_urls: Dict[str, str] = field(default_factory=lambda: DEFAULT_PROVIDER_URLS.copy())
  
      def __post_init__(self) -> None:
          """Validate configuration after initialization."""
          if not isinstance(self.port, int) or not (1 <= self.port <= 65535):
              raise ConfigurationError("Port must be between 1 and 65535")
  
          if not isinstance(self.proxy_port, int) or not (1 <= self.proxy_port <= 65535):
              raise ConfigurationError("Proxy port must be between 1 and 65535")
  
          if not isinstance(self.log_level, LogLevel):
              try:
                  self.log_level = LogLevel(self.log_level)
              except ValueError as e:
                  raise ConfigurationError(f"Invalid log level: {e}")
  
          if not isinstance(self.log_format, LogFormat):
              try:
                  self.log_format = LogFormat(self.log_format)
              except ValueError as e:
                  raise ConfigurationError(f"Invalid log format: {e}")
  
      @staticmethod
      def _load_default_prompts() -> PromptConfig:
          """Load default prompts from prompts/default.yaml."""
          default_prompts_path = Path(__file__).parent.parent.parent / "prompts" / "default.yaml"
          try:
              return PromptConfig.from_file(default_prompts_path)
          except Exception as e:
              import logging
  
              logging.warning(f"Failed to load default prompts: {e}")
              return PromptConfig()
  
      @classmethod
      def from_file(cls, config_path: Union[str, Path]) -> "Config":
          """Load configuration from a YAML file.
  
          Args:
              config_path: Path to the YAML configuration file
  
          Returns:
              Config: Configuration instance
  
          Raises:
              ConfigurationError: If the file cannot be read or parsed
          """
          try:
              with open(config_path, "r") as f:
                  config_data = yaml.safe_load(f)
  
              if not isinstance(config_data, dict):
                  raise ConfigurationError("Config file must contain a YAML dictionary")
  
              # Start with default prompts
              prompts_config = cls._load_default_prompts()
  
              # Override with prompts from config if present
              if "prompts" in config_data:
                  if isinstance(config_data["prompts"], dict):
                      prompts_config = PromptConfig(prompts=config_data.pop("prompts"))
                  elif isinstance(config_data["prompts"], str):
                      # If prompts is a string, treat it as a path to a prompts file
                      prompts_path = Path(config_data.pop("prompts"))
                      if not prompts_path.is_absolute():
                          prompts_path = Path(config_path).parent / prompts_path
                      prompts_config = PromptConfig.from_file(prompts_path)
  
              # Get provider URLs from config
              provider_urls = DEFAULT_PROVIDER_URLS.copy()
              if "provider_urls" in config_data:
                  provider_urls.update(config_data.pop("provider_urls"))
  
              return cls(
                  port=config_data.get("port", cls.port),
                  proxy_port=config_data.get("proxy_port", cls.proxy_port),
                  host=config_data.get("host", cls.host),
                  log_level=config_data.get("log_level", cls.log_level.value),
                  log_format=config_data.get("log_format", cls.log_format.value),
                  model_base_path=config_data.get("model_base_path", cls.model_base_path),
                  chat_model_n_ctx=config_data.get("chat_model_n_ctx", cls.chat_model_n_ctx),
                  chat_model_n_gpu_layers=config_data.get(
                      "chat_model_n_gpu_layers", cls.chat_model_n_gpu_layers
                  ),
                  embedding_model=config_data.get("embedding_model", cls.embedding_model),
                  db_path=config_data.get("db_path", cls.db_path),
                  vec_db_path=config_data.get("vec_db_path", cls.vec_db_path),
                  certs_dir=config_data.get("certs_dir", cls.certs_dir),
                  ca_cert=config_data.get("ca_cert", cls.ca_cert),
                  ca_key=config_data.get("ca_key", cls.ca_key),
                  server_cert=config_data.get("server_cert", cls.server_cert),
                  server_key=config_data.get("server_key", cls.server_key),
                  force_certs=config_data.get("force_certs", cls.force_certs),
                  prompts=prompts_config,
                  provider_urls=provider_urls,
                  use_update_service=config_data.get("use_update_service", cls.use_update_service),
                  update_service_url=config_data.get("update_service_url", cls.update_service_url),
              )
          except yaml.YAMLError as e:
              raise ConfigurationError(f"Failed to parse config file: {e}")
          except OSError as e:
              raise ConfigurationError(f"Failed to read config file: {e}")
  
      @classmethod
      def from_env(cls) -> "Config":
          """Load configuration from environment variables.
  
          Returns:
              Config: Configuration instance
          """
          try:
              # Start with default prompts
              config = cls(prompts=cls._load_default_prompts())
  
              if "CODEGATE_APP_PORT" in os.environ:
                  config.port = int(os.environ["CODEGATE_APP_PORT"])
              if "CODEGATE_APP_PROXY_PORT" in os.environ:
                  config.proxy_port = int(os.environ["CODEGATE_APP_PROXY_PORT"])
              if "CODEGATE_APP_HOST" in os.environ:
                  config.host = os.environ["CODEGATE_APP_HOST"]
              if "CODEGATE_APP_LOG_LEVEL" in os.environ:
                  config.log_level = LogLevel(os.environ["CODEGATE_APP_LOG_LEVEL"])
              if "CODEGATE_LOG_FORMAT" in os.environ:
                  config.log_format = LogFormat(os.environ["CODEGATE_LOG_FORMAT"])
              if "CODEGATE_PROMPTS_FILE" in os.environ:
                  config.prompts = PromptConfig.from_file(
                      os.environ["CODEGATE_PROMPTS_FILE"]
                  )  # noqa: E501
  
              # Load certificate configuration from environment
              if "CODEGATE_CERTS_DIR" in os.environ:
                  config.certs_dir = os.environ["CODEGATE_CERTS_DIR"]
              if "CODEGATE_CA_CERT" in os.environ:
                  config.ca_cert = os.environ["CODEGATE_CA_CERT"]
              if "CODEGATE_CA_KEY" in os.environ:
                  config.ca_key = os.environ["CODEGATE_CA_KEY"]
              if "CODEGATE_SERVER_CERT" in os.environ:
                  config.server_cert = os.environ["CODEGATE_SERVER_CERT"]
              if "CODEGATE_SERVER_KEY" in os.environ:
                  config.server_key = os.environ["CODEGATE_SERVER_KEY"]
              if "CODEGATE_FORCE_CERTS" in os.environ:
                  config.force_certs = cls.__bool_from_string(os.environ["CODEGATE_FORCE_CERTS"])
              if "CODEGATE_DB_PATH" in os.environ:
                  config.db_path = os.environ["CODEGATE_DB_PATH"]
              if "CODEGATE_VEC_DB_PATH" in os.environ:
                  config.vec_db_path = os.environ["CODEGATE_VEC_DB_PATH"]
              if "CODEGATE_UPDATE_SERVICE_URL" in os.environ:
                  config.update_service_url = os.environ["CODEGATE_UPDATE_SERVICE_URL"]
  
              # Load provider URLs from environment variables
              for provider in DEFAULT_PROVIDER_URLS.keys():
                  env_var = f"CODEGATE_PROVIDER_{provider.upper()}_URL"
                  if env_var in os.environ:
                      config.provider_urls[provider] = os.environ[env_var]
  
              return config
          except ValueError as e:
              raise ConfigurationError(f"Invalid environment variable value: {e}")
  
      @classmethod
      def load(
          cls,
          config_path: Optional[Union[str, Path]] = None,
          prompts_path: Optional[Union[str, Path]] = None,
          cli_port: Optional[int] = None,
          cli_proxy_port: Optional[int] = None,
          cli_host: Optional[str] = None,
          cli_log_level: Optional[str] = None,
          cli_log_format: Optional[str] = None,
          cli_provider_urls: Optional[Dict[str, str]] = None,
          model_base_path: Optional[str] = None,
          embedding_model: Optional[str] = None,
          certs_dir: Optional[str] = None,
          ca_cert: Optional[str] = None,
          ca_key: Optional[str] = None,
          server_cert: Optional[str] = None,
          server_key: Optional[str] = None,
          force_certs: Optional[bool] = None,
          db_path: Optional[str] = None,
          vec_db_path: Optional[str] = None,
          update_service_url: Optional[str] = None,
      ) -> "Config":
          """Load configuration with priority resolution.
  
          Priority order (highest to lowest):
          1. CLI arguments
          2. Environment variables
          3. Config file
          4. Default values (including default prompts from prompts/default.yaml)
  
          Args:
              config_path: Optional path to config file
              prompts_path: Optional path to prompts file
              cli_port: Optional CLI port override
              cli_proxy_port: Optional CLI proxy port override
              cli_host: Optional CLI host override
              cli_log_level: Optional CLI log level override
              cli_log_format: Optional CLI log format override
              cli_provider_urls: Optional dict of provider URLs from CLI
              model_base_path: Optional path to model base directory
              embedding_model: Optional name of the model to use for embeddings
              certs_dir: Optional path to certificates directory
              ca_cert: Optional path to CA certificate
              ca_key: Optional path to CA key
              server_cert: Optional path to server certificate
              server_key: Optional path to server key
              force_certs: Optional flag to force certificate generation
              db_path: Optional path to the main SQLite database file
              vec_db_path: Optional path to the vector SQLite database file
              update_service_url: Optional URL for the update service
  
          Returns:
              Config: Resolved configuration
  
          Raises:
              ConfigurationError: If configuration is invalid
          """
          # Start with defaults (including default prompts)
          config = cls(prompts=cls._load_default_prompts())
  
          # Load from config file if provided
          if config_path:
              try:
                  config = cls.from_file(config_path)
              except ConfigurationError as e:
                  # Log warning but continue with defaults
                  logger.warning(f"Failed to load config file: {e}")
  
          # Override with environment variables
          env_config = cls.from_env()
          if "CODEGATE_APP_PORT" in os.environ:
              config.port = env_config.port
          if "CODEGATE_APP_PROXY_PORT" in os.environ:
              config.proxy_port = env_config.proxy_port
          if "CODEGATE_APP_HOST" in os.environ:
              config.host = env_config.host
          if "CODEGATE_APP_LOG_LEVEL" in os.environ:
              config.log_level = env_config.log_level
          if "CODEGATE_LOG_FORMAT" in os.environ:
              config.log_format = env_config.log_format
          if "CODEGATE_PROMPTS_FILE" in os.environ:
              config.prompts = env_config.prompts
          if "CODEGATE_MODEL_BASE_PATH" in os.environ:
              config.model_base_path = env_config.model_base_path
          if "CODEGATE_EMBEDDING_MODEL" in os.environ:
              config.embedding_model = env_config.embedding_model
          if "CODEGATE_CERTS_DIR" in os.environ:
              config.certs_dir = env_config.certs_dir
          if "CODEGATE_CA_CERT" in os.environ:
              config.ca_cert = env_config.ca_cert
          if "CODEGATE_CA_KEY" in os.environ:
              config.ca_key = env_config.ca_key
          if "CODEGATE_SERVER_CERT" in os.environ:
              config.server_cert = env_config.server_cert
          if "CODEGATE_SERVER_KEY" in os.environ:
              config.server_key = env_config.server_key
          if "CODEGATE_FORCE_CERTS" in os.environ:
              config.force_certs = env_config.force_certs
          if "CODEGATE_DB_PATH" in os.environ:
              config.db_path = env_config.db_path
          if "CODEGATE_VEC_DB_PATH" in os.environ:
              config.vec_db_path = env_config.vec_db_path
          if "CODEGATE_UPDATE_SERVICE_URL" in os.environ:
              config.update_service_url = env_config.update_service_url
  
          # Override provider URLs from environment
          for provider, url in env_config.provider_urls.items():
              config.provider_urls[provider] = url
  
          # Override with CLI arguments
          if cli_port is not None:
              config.port = cli_port
          if cli_proxy_port is not None:
              config.proxy_port = cli_proxy_port
          if cli_host is not None:
              config.host = cli_host
          if cli_log_level is not None:
              config.log_level = LogLevel(cli_log_level)
          if cli_log_format is not None:
              config.log_format = LogFormat(cli_log_format)
          if prompts_path is not None:
              config.prompts = PromptConfig.from_file(prompts_path)
          if cli_provider_urls is not None:
              config.provider_urls.update(cli_provider_urls)
          if model_base_path is not None:
              config.model_base_path = model_base_path
          if embedding_model is not None:
              config.embedding_model = embedding_model
          if certs_dir is not None:
              config.certs_dir = certs_dir
          if ca_cert is not None:
              config.ca_cert = ca_cert
          if ca_key is not None:
              config.ca_key = ca_key
          if server_cert is not None:
              config.server_cert = server_cert
          if server_key is not None:
              config.server_key = server_key
          if db_path is not None:
              config.db_path = db_path
          if vec_db_path is not None:
              config.vec_db_path = vec_db_path
          if force_certs is not None:
              config.force_certs = force_certs
          if update_service_url is not None:
              config.update_service_url = update_service_url
  
          # Set the __config class attribute
          Config.__config = config
  
          return config
  
      @classmethod
      def get_config(cls) -> "Config":
          return cls.__config
  
      @staticmethod
      def __bool_from_string(raw_value) -> bool:
          return raw_value.lower() == "true"
  