ccproxy.auth.storage.base

ccproxy.auth.storage.base

Abstract base class for token storage.

TokenStorage

Bases: ABC, Generic[CredentialsT]

Abstract interface for token storage operations.

This is a generic interface that can work with any credential type that extends BaseModel (e.g., ClaudeCredentials, OpenAICredentials).

load abstractmethod async

load()

Load credentials from storage.

Returns:

Type	Description
CredentialsT | None	Parsed credentials if found and valid, None otherwise

Source code in ccproxy/auth/storage/base.py

@abstractmethod
async def load(self) -> CredentialsT | None:
    """Load credentials from storage.

    Returns:
        Parsed credentials if found and valid, None otherwise
    """
    pass

save abstractmethod async

save(credentials)

Save credentials to storage.

Parameters:

Name	Type	Description	Default
credentials	CredentialsT	Credentials to save	required

Returns:

Type	Description
bool	True if saved successfully, False otherwise

Source code in ccproxy/auth/storage/base.py

@abstractmethod
async def save(self, credentials: CredentialsT) -> bool:
    """Save credentials to storage.

    Args:
        credentials: Credentials to save

    Returns:
        True if saved successfully, False otherwise
    """
    pass

exists abstractmethod async

exists()

Check if credentials exist in storage.

Returns:

Type	Description
bool	True if credentials exist, False otherwise

Source code in ccproxy/auth/storage/base.py

@abstractmethod
async def exists(self) -> bool:
    """Check if credentials exist in storage.

    Returns:
        True if credentials exist, False otherwise
    """
    pass

delete abstractmethod async

delete()

Delete credentials from storage.

Returns:

Type	Description
bool	True if deleted successfully, False otherwise

Source code in ccproxy/auth/storage/base.py

@abstractmethod
async def delete(self) -> bool:
    """Delete credentials from storage.

    Returns:
        True if deleted successfully, False otherwise
    """
    pass

get_location abstractmethod

get_location()

Get the storage location description.

Returns:

Type	Description
str	Human-readable description of where credentials are stored

Source code in ccproxy/auth/storage/base.py

@abstractmethod
def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Human-readable description of where credentials are stored
    """
    pass

BaseJsonStorage

BaseJsonStorage(file_path, enable_backups=True)

Bases: TokenStorage[CredentialsT], Generic[CredentialsT]

Base class for JSON file storage implementations.

This class provides common JSON read/write operations with error handling, atomic writes, and proper permission management.

This is a generic class that can work with any credential type.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to JSON file for storage	required
enable_backups	bool	Whether to create backups before overwriting	True

Source code in ccproxy/auth/storage/base.py

def __init__(self, file_path: Path, enable_backups: bool = True):
    """Initialize JSON storage.

    Args:
        file_path: Path to JSON file for storage
        enable_backups: Whether to create backups before overwriting
    """
    self.file_path = file_path
    self.enable_backups = enable_backups

exists async

exists()

Check if credentials file exists.

Returns:

Type	Description
bool	True if file exists, False otherwise

Source code in ccproxy/auth/storage/base.py

async def exists(self) -> bool:
    """Check if credentials file exists.

    Returns:
        True if file exists, False otherwise
    """
    # Run file system check in thread pool for consistency
    file_exists = await asyncio.to_thread(
        lambda: self.file_path.exists() and self.file_path.is_file()
    )

    logger.debug(
        "auth_file_existence_check",
        file_path=str(self.file_path),
        exists=file_exists,
        category="auth",
    )

    return file_exists

delete async

delete()

Delete credentials file.

Returns:

Type	Description
bool	True if deleted successfully, False if file didn't exist

Raises:

Type	Description
CredentialsStorageError	If file cannot be deleted

Source code in ccproxy/auth/storage/base.py

async def delete(self) -> bool:
    """Delete credentials file.

    Returns:
        True if deleted successfully, False if file didn't exist

    Raises:
        CredentialsStorageError: If file cannot be deleted
    """
    try:
        if await self.exists():
            await asyncio.to_thread(self.file_path.unlink)
            logger.debug("file_deleted", path=str(self.file_path))
            return True
        return False

    except PermissionError as e:
        logger.error(
            "permission_denied",
            path=str(self.file_path),
            error=str(e),
            exc_info=e,
        )
        raise CredentialsStorageError(f"Permission denied: {self.file_path}") from e

    except OSError as e:
        logger.error(
            "file_delete_error",
            path=str(self.file_path),
            error=str(e),
            exc_info=e,
        )
        raise CredentialsStorageError(
            f"Error deleting {self.file_path}: {e}"
        ) from e

get_location

get_location()

Get the storage location description.

Returns:

Type	Description
str	Path to the JSON file

Source code in ccproxy/auth/storage/base.py

def get_location(self) -> str:
    """Get the storage location description.

    Returns:
        Path to the JSON file
    """
    return str(self.file_path)