ccproxy.plugins.oauth_claude.storage¶

ccproxy.plugins.oauth_claude.storage ¶

Token storage for Claude OAuth plugin.

ClaudeOAuthStorage ¶

ClaudeOAuthStorage(storage_path=None)

Bases: BaseJsonStorage[ClaudeCredentials]

Claude OAuth-specific token storage implementation.

Parameters:

Name	Type	Description	Default
`storage_path`	`Path \| None`	Path to storage file	`None`

Source code in ccproxy/plugins/oauth_claude/storage.py

def __init__(self, storage_path: Path | None = None):
    """Initialize Claude OAuth token storage.

    Args:
        storage_path: Path to storage file
    """
    if storage_path is None:
        # Default to standard Claude credentials location
        storage_path = Path.home() / ".claude" / ".credentials.json"

    super().__init__(storage_path)
    self.provider_name = "claude-api"

save `async` ¶

save(credentials)

Save Claude credentials.

Parameters:

Name	Type	Description	Default
`credentials`	`ClaudeCredentials`	Claude credentials to save	required

Returns:

Type	Description
`bool`	True if saved successfully, False otherwise

Source code in ccproxy/plugins/oauth_claude/storage.py

async def save(self, credentials: ClaudeCredentials) -> bool:
    """Save Claude credentials.

    Args:
        credentials: Claude credentials to save

    Returns:
        True if saved successfully, False otherwise
    """
    try:
        # Convert to dict for storage (uses by_alias=True by default)
        data = credentials.model_dump(mode="json", exclude_none=True)

        # Use parent class's atomic write with backup
        await self._write_json(data)

        logger.debug(
            "claude_oauth_credentials_saved",
            has_oauth=bool(credentials.claude_ai_oauth),
            storage_path=str(self.file_path),
            category="auth",
        )
        return True
    except Exception as e:
        logger.error(
            "claude_oauth_save_failed", error=str(e), exc_info=e, category="auth"
        )
        return False

load `async` ¶

load()

Load Claude credentials.

Returns:

Type	Description
`ClaudeCredentials \| None`	Stored credentials or None

Source code in ccproxy/plugins/oauth_claude/storage.py

async def load(self) -> ClaudeCredentials | None:
    """Load Claude credentials.

    Returns:
        Stored credentials or None
    """
    try:
        # Use parent class's read method
        data = await self._read_json()
        if not data:
            return None

        credentials = ClaudeCredentials.model_validate(data)
        logger.debug(
            "claude_oauth_credentials_loaded",
            has_oauth=bool(credentials.claude_ai_oauth),
            category="auth",
        )
        return credentials
    except Exception as e:
        logger.error(
            "claude_oauth_credentials_load_error",
            error=str(e),
            exc_info=e,
            category="auth",
        )
        return None

ClaudeProfileStorage ¶

ClaudeProfileStorage(storage_path=None)

Claude profile storage implementation for .account.json.

Parameters:

Name	Type	Description	Default
`storage_path`	`Path \| None`	Path to storage file	`None`

Source code in ccproxy/plugins/oauth_claude/storage.py

def __init__(self, storage_path: Path | None = None):
    """Initialize Claude profile storage.

    Args:
        storage_path: Path to storage file
    """
    if storage_path is None:
        # Default to standard Claude account location
        storage_path = Path.home() / ".claude" / ".account.json"

    self.file_path = storage_path

save_profile `async` ¶

save_profile(profile_data)

Save Claude profile data.

Parameters:

Name	Type	Description	Default
`profile_data`	`dict[str, Any]`	Raw profile data from API	required

Returns:

Type	Description
`bool`	True if saved successfully, False otherwise

Source code in ccproxy/plugins/oauth_claude/storage.py

async def save_profile(self, profile_data: dict[str, Any]) -> bool:
    """Save Claude profile data.

    Args:
        profile_data: Raw profile data from API

    Returns:
        True if saved successfully, False otherwise
    """
    try:
        # Write the raw profile data
        await self._write_json(profile_data)

        # Extract key info for logging
        account = profile_data.get("account", {})
        logger.info(
            "claude_profile_saved",
            account_id=account.get("uuid"),
            email=account.get("email"),
            has_claude_pro=account.get("has_claude_pro"),
            has_claude_max=account.get("has_claude_max"),
            storage_path=str(self.file_path),
            category="auth",
        )
        return True
    except Exception as e:
        logger.error(
            "claude_profile_save_failed",
            error=str(e),
            exc_info=e,
            category="auth",
        )
        return False

load_profile `async` ¶

load_profile()

Load Claude profile.

Returns:

Type	Description
`ClaudeProfileInfo \| None`	ClaudeProfileInfo or None if not found

Source code in ccproxy/plugins/oauth_claude/storage.py

async def load_profile(self) -> ClaudeProfileInfo | None:
    """Load Claude profile.

    Returns:
        ClaudeProfileInfo or None if not found
    """
    try:
        data = await self._read_json()
        if not data:
            return None

        profile = ClaudeProfileInfo.from_api_response(data)
        logger.debug(
            "claude_profile_loaded",
            account_id=profile.account_id,
            email=profile.email,
            category="auth",
        )
        return profile
    except Exception as e:
        logger.error(
            "claude_profile_load_error",
            error=str(e),
            exc_info=e,
            category="auth",
        )
        return None

ccproxy.plugins.oauth_claude.storage¶