Skip to content

ccproxy.auth.storage.generic

ccproxy.auth.storage.generic

Generic storage implementation using Pydantic validation.

GenericJsonStorage 

GenericJsonStorage(file_path, model_class)

Bases: BaseJsonStorage[T]

Generic storage implementation using Pydantic validation.

This replaces provider-specific storage classes with a single implementation that handles any Pydantic model.

Parameters:

Name Type Description Default
file_path Path

Path to JSON file

 required
model_class type[T]

Pydantic model class for validation

 required
Source code in ccproxy/auth/storage/generic.py 
26
27
28
29
30
31
32
33
34
35
def __init__(self, file_path: Path, model_class: type[T]):
    """Initialize generic storage.

    Args:
        file_path: Path to JSON file
        model_class: Pydantic model class for validation
    """
    super().__init__(file_path)
    self.model_class = model_class
    self.type_adapter = TypeAdapter(model_class)

load async 

load()

Load and validate credentials with Pydantic.

Returns:

Type Description
T | None

Validated model instance or None if file doesn't exist
Source code in ccproxy/auth/storage/generic.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def load(self) -> T | None:
    """Load and validate credentials with Pydantic.

    Returns:
        Validated model instance or None if file doesn't exist
    """
    try:
        data = await self._read_json()
    except FileNotFoundError:
        # File doesn't exist - this is normal for uninitialized credentials
        logger.debug(
            "credential_file_not_found",
            path=str(self.file_path),
            category="auth",
        )
        return None
    except Exception as e:
        # Handle JSON decode errors and other file read issues with clear warning
        error_type = type(e).__name__
        logger.warning(
            "credential_file_read_failed",
            error_type=error_type,
            error=str(e),
            exc_info=e,
            path=str(self.file_path),
            category="auth",
        )
        return None

    if not data:
        return None

    try:
        # Pydantic handles all validation and conversion
        return self.type_adapter.validate_python(data)
    except Exception as e:
        # Log validation errors with clean warning (not error)
        error_type = type(e).__name__
        logger.warning(
            "credential_validation_failed",
            error_type=error_type,
            error=str(e),
            exc_info=e,
            model=self.model_class.__name__,
            path=str(self.file_path),
            category="auth",
        )
        return None

save async 

save(obj)

Save model using Pydantic serialization.

Parameters:

Name Type Description Default
obj T

Pydantic model instance to save

 required

Returns:

Type Description
bool

True if saved successfully
Source code in ccproxy/auth/storage/generic.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def save(self, obj: T) -> bool:
    """Save model using Pydantic serialization.

    Args:
        obj: Pydantic model instance to save

    Returns:
        True if saved successfully
    """
    try:
        # Preserve original JSON structure using aliases
        # Use dump_python without mode="json" to get actual values
        data = self.type_adapter.dump_python(
            obj,
            by_alias=True,  # Use field aliases from original models
            exclude_none=True,
        )
        # Convert SecretStr values to their actual values
        data = self._unmask_secrets(data)
        await self._write_json(data)
        return True
    except Exception as e:
        logger.error(
            "Failed to save credentials",
            error=str(e),
            exc_info=e,
            model=self.model_class.__name__,
        )
        return False