ccproxy.config.utils

ccproxy.config.utils

Configuration utilities - constants, validators, discovery, and scheduler.

ConfigValidationError

Bases: Exception

Configuration validation error.

SchedulerSettings

Bases: BaseSettings

Configuration settings for the unified scheduler system.

Controls global scheduler behavior and individual task configurations. Settings can be configured via environment variables with SCHEDULER__ prefix.

validate_host

validate_host(host)

Validate host address.

Parameters:

Name	Type	Description	Default
host	str	Host address to validate	required

Returns:

Type	Description
str	The validated host address

Raises:

Type	Description
ConfigValidationError	If host is invalid

Source code in ccproxy/config/utils.py

def validate_host(host: str) -> str:
    """Validate host address.

    Args:
        host: Host address to validate

    Returns:
        The validated host address

    Raises:
        ConfigValidationError: If host is invalid
    """
    if not host:
        raise ConfigValidationError("Host cannot be empty")

    # Allow localhost, IP addresses, and domain names
    if host in ["localhost", "0.0.0.0", "127.0.0.1"]:
        return host

    # Basic IP address validation
    if re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", host):
        parts = host.split(".")
        if all(0 <= int(part) <= 255 for part in parts):
            return host
        raise ConfigValidationError(f"Invalid IP address: {host}")

    # Basic domain name validation
    if re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", host):
        return host

    return host  # Allow other formats for flexibility

validate_port

validate_port(port)

Validate port number.

Parameters:

Name	Type	Description	Default
port	int | str	Port number to validate	required

Returns:

Type	Description
int	The validated port number

Raises:

Type	Description
ConfigValidationError	If port is invalid

Source code in ccproxy/config/utils.py

def validate_port(port: int | str) -> int:
    """Validate port number.

    Args:
        port: Port number to validate

    Returns:
        The validated port number

    Raises:
        ConfigValidationError: If port is invalid
    """
    if isinstance(port, str):
        try:
            port = int(port)
        except ValueError as e:
            raise ConfigValidationError(f"Port must be a valid integer: {port}") from e

    if not isinstance(port, int):
        raise ConfigValidationError(f"Port must be an integer: {port}")

    if port < 1 or port > 65535:
        raise ConfigValidationError(f"Port must be between 1 and 65535: {port}")

    return port

validate_url

validate_url(url)

Validate URL format.

Parameters:

Name	Type	Description	Default
url	str	URL to validate	required

Returns:

Type	Description
str	The validated URL

Raises:

Type	Description
ConfigValidationError	If URL is invalid

Source code in ccproxy/config/utils.py

def validate_url(url: str) -> str:
    """Validate URL format.

    Args:
        url: URL to validate

    Returns:
        The validated URL

    Raises:
        ConfigValidationError: If URL is invalid
    """
    if not url:
        raise ConfigValidationError("URL cannot be empty")

    try:
        result = urlparse(url)
        if not result.scheme or not result.netloc:
            raise ConfigValidationError(f"Invalid URL format: {url}")
    except Exception as e:
        raise ConfigValidationError(f"Invalid URL: {url}") from e

    return url

validate_path

validate_path(path)

Validate file path.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to validate	required

Returns:

Type	Description
Path	The validated Path object

Raises:

Type	Description
ConfigValidationError	If path is invalid

Source code in ccproxy/config/utils.py

def validate_path(path: str | Path) -> Path:
    """Validate file path.

    Args:
        path: Path to validate

    Returns:
        The validated Path object

    Raises:
        ConfigValidationError: If path is invalid
    """
    if isinstance(path, str):
        path = Path(path)

    if not isinstance(path, Path):
        raise ConfigValidationError(f"Path must be a string or Path object: {path}")

    return path

validate_log_level

validate_log_level(level)

Validate log level.

Parameters:

Name	Type	Description	Default
level	str	Log level to validate	required

Returns:

Type	Description
str	The validated log level

Raises:

Type	Description
ConfigValidationError	If log level is invalid

Source code in ccproxy/config/utils.py

def validate_log_level(level: str) -> str:
    """Validate log level.

    Args:
        level: Log level to validate

    Returns:
        The validated log level

    Raises:
        ConfigValidationError: If log level is invalid
    """
    valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
    level = level.upper()

    if level not in valid_levels:
        raise ConfigValidationError(
            f"Invalid log level: {level}. Must be one of: {valid_levels}"
        )

    return level

validate_cors_origins

validate_cors_origins(origins)

Validate CORS origins.

Parameters:

Name	Type	Description	Default
origins	list[str]	List of origin URLs to validate	required

Returns:

Type	Description
list[str]	The validated list of origins

Raises:

Type	Description
ConfigValidationError	If any origin is invalid

Source code in ccproxy/config/utils.py

def validate_cors_origins(origins: list[str]) -> list[str]:
    """Validate CORS origins.

    Args:
        origins: List of origin URLs to validate

    Returns:
        The validated list of origins

    Raises:
        ConfigValidationError: If any origin is invalid
    """
    if not isinstance(origins, list):
        raise ConfigValidationError("CORS origins must be a list")

    validated_origins = []
    for origin in origins:
        if origin == "*":
            validated_origins.append(origin)
        else:
            validated_origins.append(validate_url(origin))

    return validated_origins

validate_timeout

validate_timeout(timeout)

Validate timeout value.

Parameters:

Name	Type	Description	Default
timeout	int | float	Timeout value to validate	required

Returns:

Type	Description
int | float	The validated timeout value

Raises:

Type	Description
ConfigValidationError	If timeout is invalid

Source code in ccproxy/config/utils.py

def validate_timeout(timeout: int | float) -> int | float:
    """Validate timeout value.

    Args:
        timeout: Timeout value to validate

    Returns:
        The validated timeout value

    Raises:
        ConfigValidationError: If timeout is invalid
    """
    if not isinstance(timeout, int | float):
        raise ConfigValidationError(f"Timeout must be a number: {timeout}")

    if timeout <= 0:
        raise ConfigValidationError(f"Timeout must be positive: {timeout}")

    return timeout

validate_config_dict

validate_config_dict(config)

Validate configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary to validate	required

Returns:

Type	Description
dict[str, Any]	The validated configuration dictionary

Raises:

Type	Description
ConfigValidationError	If configuration is invalid

Source code in ccproxy/config/utils.py

def validate_config_dict(config: dict[str, Any]) -> dict[str, Any]:
    """Validate configuration dictionary.

    Args:
        config: Configuration dictionary to validate

    Returns:
        The validated configuration dictionary

    Raises:
        ConfigValidationError: If configuration is invalid
    """
    if not isinstance(config, dict):
        raise ConfigValidationError("Configuration must be a dictionary")

    validated_config: dict[str, Any] = {}

    # Validate specific fields if present
    if "host" in config:
        validated_config["host"] = validate_host(config["host"])

    if "port" in config:
        validated_config["port"] = validate_port(config["port"])

    if "target_url" in config:
        validated_config["target_url"] = validate_url(config["target_url"])

    if "log_level" in config:
        validated_config["log_level"] = validate_log_level(config["log_level"])

    if "cors_origins" in config:
        validated_config["cors_origins"] = validate_cors_origins(config["cors_origins"])

    if "timeout" in config:
        validated_config["timeout"] = validate_timeout(config["timeout"])

    # Copy other fields without validation
    for key, value in config.items():
        if key not in validated_config:
            validated_config[key] = value

    return validated_config

find_toml_config_file

find_toml_config_file()

Find the TOML configuration file for ccproxy.

Searches in the following order: 1. .ccproxy.toml in current directory 2. ccproxy.toml in git repository root (if in a git repo) 3. config.toml in XDG_CONFIG_HOME/ccproxy/

Source code in ccproxy/config/utils.py

def find_toml_config_file() -> Path | None:
    """Find the TOML configuration file for ccproxy.

    Searches in the following order:
    1. .ccproxy.toml in current directory
    2. ccproxy.toml in git repository root (if in a git repo)
    3. config.toml in XDG_CONFIG_HOME/ccproxy/
    """
    # Check current directory first
    candidates = [
        Path(".ccproxy.toml").resolve(),
        Path("ccproxy.toml").resolve(),
    ]

    # Check git repo root
    git_root = find_git_root()
    if git_root:
        candidates.extend(
            [
                git_root / ".ccproxy.toml",
                git_root / "ccproxy.toml",
            ]
        )

    # Check XDG config directory
    config_dir = get_ccproxy_config_dir()
    candidates.append(config_dir / "config.toml")

    # Return first existing file
    for candidate in candidates:
        if candidate.exists() and candidate.is_file():
            return candidate

    return None

find_git_root

find_git_root(path=None)

Find the root directory of a git repository.

Source code in ccproxy/config/utils.py

def find_git_root(path: Path | None = None) -> Path | None:
    """Find the root directory of a git repository."""
    if path is None:
        path = Path.cwd()

    try:
        result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            cwd=path,
            capture_output=True,
            text=True,
            check=True,
        )
        return Path(result.stdout.strip())
    except (subprocess.CalledProcessError, FileNotFoundError):
        return None

get_ccproxy_config_dir

get_ccproxy_config_dir()

Get the ccproxy configuration directory.

Returns:

Type	Description
Path	Path to the ccproxy configuration directory within XDG_CONFIG_HOME.

Source code in ccproxy/config/utils.py

def get_ccproxy_config_dir() -> Path:
    """Get the ccproxy configuration directory.

    Returns:
        Path to the ccproxy configuration directory within XDG_CONFIG_HOME.
    """
    return get_xdg_config_home() / "ccproxy"

get_claude_cli_config_dir

get_claude_cli_config_dir()

Get the Claude CLI configuration directory.

Returns:

Type	Description
Path	Path to the Claude CLI configuration directory within XDG_CONFIG_HOME.

Source code in ccproxy/config/utils.py

def get_claude_cli_config_dir() -> Path:
    """Get the Claude CLI configuration directory.

    Returns:
        Path to the Claude CLI configuration directory within XDG_CONFIG_HOME.
    """
    return get_xdg_config_home() / "claude"

get_claude_docker_home_dir

get_claude_docker_home_dir()

Get the Claude Docker home directory.

Returns:

Type	Description
Path	Path to the Claude Docker home directory within XDG_DATA_HOME.

Source code in ccproxy/config/utils.py

def get_claude_docker_home_dir() -> Path:
    """Get the Claude Docker home directory.

    Returns:
        Path to the Claude Docker home directory within XDG_DATA_HOME.
    """
    return get_ccproxy_config_dir() / "home"

get_ccproxy_cache_dir

get_ccproxy_cache_dir()

Get the ccproxy cache directory.

Returns:

Type	Description
Path	Path to the ccproxy cache directory within XDG_CACHE_HOME.

Source code in ccproxy/config/utils.py

def get_ccproxy_cache_dir() -> Path:
    """Get the ccproxy cache directory.

    Returns:
        Path to the ccproxy cache directory within XDG_CACHE_HOME.
    """
    return get_xdg_cache_home() / "ccproxy"