Skip to content

infrastructure.cache.cache_keys

src.infrastructure.cache.cache_keys

Cache key construction utilities.

This module provides centralized cache key construction to ensure consistency across all cache operations. All keys follow the pattern: {prefix}:{domain}:{resource}:{id}

Usage

from src.core.container import get_settings from src.infrastructure.cache.cache_keys import CacheKeys

settings = get_settings() keys = CacheKeys(prefix=settings.cache_key_prefix)

Construct keys

user_key = keys.user(user_id) provider_key = keys.provider_connection(connection_id)

Classes

CacheKeys dataclass

Centralized cache key construction utilities.

Ensures consistent key patterns across all cache operations. All keys follow hierarchical structure: {prefix}:{domain}:{resource}:{id}

Attributes:

Name Type Description
prefix str

Cache key prefix (typically "dashtam").
Example

keys = CacheKeys(prefix="dashtam") key = keys.user(user_id) # "dashtam:user:{user_id}"

Source code in src/infrastructure/cache/cache_keys.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 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
 85
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@dataclass
class CacheKeys:
    """Centralized cache key construction utilities.

    Ensures consistent key patterns across all cache operations.
    All keys follow hierarchical structure: {prefix}:{domain}:{resource}:{id}

    Attributes:
        prefix: Cache key prefix (typically "dashtam").

    Example:
        keys = CacheKeys(prefix="dashtam")
        key = keys.user(user_id)  # "dashtam:user:{user_id}"
    """

    prefix: str

    def user(self, user_id: UUID) -> str:
        """User data cache key.

        Pattern: {prefix}:user:{user_id}

        Args:
            user_id: User UUID.

        Returns:
            Cache key string.

        Example:
            "dashtam:user:123e4567-e89b-12d3-a456-426614174000"
        """
        return f"{self.prefix}:user:{user_id}"

    def provider_connection(self, connection_id: UUID) -> str:
        """Provider connection cache key.

        Pattern: {prefix}:provider:conn:{connection_id}

        Args:
            connection_id: ProviderConnection UUID.

        Returns:
            Cache key string.

        Example:
            "dashtam:provider:conn:456e7890-e89b-12d3-a456-426614174001"
        """
        return f"{self.prefix}:provider:conn:{connection_id}"

    def schwab_accounts(self, user_id: UUID) -> str:
        """Schwab accounts cache key.

        Pattern: {prefix}:schwab:accounts:{user_id}

        Args:
            user_id: User UUID who owns the accounts.

        Returns:
            Cache key string.

        Example:
            "dashtam:schwab:accounts:123e4567-e89b-12d3-a456-426614174000"
        """
        return f"{self.prefix}:schwab:accounts:{user_id}"

    def schwab_transactions(
        self,
        account_id: UUID,
        start_date: date,
        end_date: date,
    ) -> str:
        """Schwab transactions cache key.

        Pattern: {prefix}:schwab:tx:{account_id}:{start_date}:{end_date}

        Args:
            account_id: Account UUID.
            start_date: Transaction query start date.
            end_date: Transaction query end date.

        Returns:
            Cache key string.

        Example:
            "dashtam:schwab:tx:789e0123-e89b-12d3-a456-426614174002:2025-01-01:2025-01-31"
        """
        return f"{self.prefix}:schwab:tx:{account_id}:{start_date}:{end_date}"

    def account_list(self, user_id: UUID) -> str:
        """Account list cache key.

        Pattern: {prefix}:accounts:user:{user_id}

        Args:
            user_id: User UUID who owns the accounts.

        Returns:
            Cache key string.

        Example:
            "dashtam:accounts:user:123e4567-e89b-12d3-a456-426614174000"
        """
        return f"{self.prefix}:accounts:user:{user_id}"

    def security_global_version(self) -> str:
        """Security global token version cache key.

        Pattern: {prefix}:security:global_version

        Returns:
            Cache key string.

        Example:
            "dashtam:security:global_version"

        Note:
            Used for token breach rotation (F1.3b).
            Caches global_min_token_version for fast JWT validation.
        """
        return f"{self.prefix}:security:global_version"

    def security_user_version(self, user_id: UUID) -> str:
        """Security user token version cache key.

        Pattern: {prefix}:security:user_version:{user_id}

        Args:
            user_id: User UUID.

        Returns:
            Cache key string.

        Example:
            "dashtam:security:user_version:123e4567-e89b-12d3-a456-426614174000"

        Note:
            Used for token breach rotation (F1.3b).
            Caches user.min_token_version for fast JWT validation.
        """
        return f"{self.prefix}:security:user_version:{user_id}"

    def namespace_from_key(self, key: str) -> str:
        """Extract cache namespace from key for metrics tracking.

        Args:
            key: Cache key string.

        Returns:
            Namespace string (e.g., "user", "provider", "accounts").

        Example:
            key = "dashtam:user:123e4567-e89b-12d3-a456-426614174000"
            namespace = keys.namespace_from_key(key)  # "user"
        """
        parts = key.split(":")
        if len(parts) >= 2:
            return parts[1]
        return "unknown"
Functions
user 
user(user_id: UUID) -> str

User data cache key.

Pattern: {prefix}:user:{user_id}

Parameters:

Name Type Description Default
user_id UUID

User UUID.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:user:123e4567-e89b-12d3-a456-426614174000"

Source code in src/infrastructure/cache/cache_keys.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def user(self, user_id: UUID) -> str:
    """User data cache key.

    Pattern: {prefix}:user:{user_id}

    Args:
        user_id: User UUID.

    Returns:
        Cache key string.

    Example:
        "dashtam:user:123e4567-e89b-12d3-a456-426614174000"
    """
    return f"{self.prefix}:user:{user_id}"
provider_connection 
provider_connection(connection_id: UUID) -> str

Provider connection cache key.

Pattern: {prefix}:provider:conn:{connection_id}

Parameters:

Name Type Description Default
connection_id UUID

ProviderConnection UUID.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:provider:conn:456e7890-e89b-12d3-a456-426614174001"

Source code in src/infrastructure/cache/cache_keys.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def provider_connection(self, connection_id: UUID) -> str:
    """Provider connection cache key.

    Pattern: {prefix}:provider:conn:{connection_id}

    Args:
        connection_id: ProviderConnection UUID.

    Returns:
        Cache key string.

    Example:
        "dashtam:provider:conn:456e7890-e89b-12d3-a456-426614174001"
    """
    return f"{self.prefix}:provider:conn:{connection_id}"
schwab_accounts 
schwab_accounts(user_id: UUID) -> str

Schwab accounts cache key.

Pattern: {prefix}:schwab:accounts:{user_id}

Parameters:

Name Type Description Default
user_id UUID

User UUID who owns the accounts.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:schwab:accounts:123e4567-e89b-12d3-a456-426614174000"

Source code in src/infrastructure/cache/cache_keys.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def schwab_accounts(self, user_id: UUID) -> str:
    """Schwab accounts cache key.

    Pattern: {prefix}:schwab:accounts:{user_id}

    Args:
        user_id: User UUID who owns the accounts.

    Returns:
        Cache key string.

    Example:
        "dashtam:schwab:accounts:123e4567-e89b-12d3-a456-426614174000"
    """
    return f"{self.prefix}:schwab:accounts:{user_id}"
schwab_transactions 
schwab_transactions(
    account_id: UUID, start_date: date, end_date: date
) -> str

Schwab transactions cache key.

Pattern: {prefix}:schwab:tx:{account_id}:{start_date}:{end_date}

Parameters:

Name Type Description Default
account_id UUID

Account UUID.

 required
start_date date

Transaction query start date.

 required
end_date date

Transaction query end date.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:schwab:tx:789e0123-e89b-12d3-a456-426614174002:2025-01-01:2025-01-31"

Source code in src/infrastructure/cache/cache_keys.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def schwab_transactions(
    self,
    account_id: UUID,
    start_date: date,
    end_date: date,
) -> str:
    """Schwab transactions cache key.

    Pattern: {prefix}:schwab:tx:{account_id}:{start_date}:{end_date}

    Args:
        account_id: Account UUID.
        start_date: Transaction query start date.
        end_date: Transaction query end date.

    Returns:
        Cache key string.

    Example:
        "dashtam:schwab:tx:789e0123-e89b-12d3-a456-426614174002:2025-01-01:2025-01-31"
    """
    return f"{self.prefix}:schwab:tx:{account_id}:{start_date}:{end_date}"
account_list 
account_list(user_id: UUID) -> str

Account list cache key.

Pattern: {prefix}:accounts:user:{user_id}

Parameters:

Name Type Description Default
user_id UUID

User UUID who owns the accounts.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:accounts:user:123e4567-e89b-12d3-a456-426614174000"

Source code in src/infrastructure/cache/cache_keys.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def account_list(self, user_id: UUID) -> str:
    """Account list cache key.

    Pattern: {prefix}:accounts:user:{user_id}

    Args:
        user_id: User UUID who owns the accounts.

    Returns:
        Cache key string.

    Example:
        "dashtam:accounts:user:123e4567-e89b-12d3-a456-426614174000"
    """
    return f"{self.prefix}:accounts:user:{user_id}"
security_global_version 
security_global_version() -> str

Security global token version cache key.

Pattern: {prefix}:security:global_version

Returns:

Type Description
str

Cache key string.
Example

"dashtam:security:global_version"

Note

Used for token breach rotation (F1.3b). Caches global_min_token_version for fast JWT validation.

Source code in src/infrastructure/cache/cache_keys.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def security_global_version(self) -> str:
    """Security global token version cache key.

    Pattern: {prefix}:security:global_version

    Returns:
        Cache key string.

    Example:
        "dashtam:security:global_version"

    Note:
        Used for token breach rotation (F1.3b).
        Caches global_min_token_version for fast JWT validation.
    """
    return f"{self.prefix}:security:global_version"
security_user_version 
security_user_version(user_id: UUID) -> str

Security user token version cache key.

Pattern: {prefix}:security:user_version:{user_id}

Parameters:

Name Type Description Default
user_id UUID

User UUID.

 required

Returns:

Type Description
str

Cache key string.
Example

"dashtam:security:user_version:123e4567-e89b-12d3-a456-426614174000"

Note

Used for token breach rotation (F1.3b). Caches user.min_token_version for fast JWT validation.

Source code in src/infrastructure/cache/cache_keys.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def security_user_version(self, user_id: UUID) -> str:
    """Security user token version cache key.

    Pattern: {prefix}:security:user_version:{user_id}

    Args:
        user_id: User UUID.

    Returns:
        Cache key string.

    Example:
        "dashtam:security:user_version:123e4567-e89b-12d3-a456-426614174000"

    Note:
        Used for token breach rotation (F1.3b).
        Caches user.min_token_version for fast JWT validation.
    """
    return f"{self.prefix}:security:user_version:{user_id}"
namespace_from_key 
namespace_from_key(key: str) -> str

Extract cache namespace from key for metrics tracking.

Parameters:

Name Type Description Default
key str

Cache key string.

 required

Returns:

Type Description
str

Namespace string (e.g., "user", "provider", "accounts").
Example

key = "dashtam:user:123e4567-e89b-12d3-a456-426614174000" namespace = keys.namespace_from_key(key) # "user"

Source code in src/infrastructure/cache/cache_keys.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def namespace_from_key(self, key: str) -> str:
    """Extract cache namespace from key for metrics tracking.

    Args:
        key: Cache key string.

    Returns:
        Namespace string (e.g., "user", "provider", "accounts").

    Example:
        key = "dashtam:user:123e4567-e89b-12d3-a456-426614174000"
        namespace = keys.namespace_from_key(key)  # "user"
    """
    parts = key.split(":")
    if len(parts) >= 2:
        return parts[1]
    return "unknown"