Skip to content

domain.protocols.cache_keys_protocol

src.domain.protocols.cache_keys_protocol

Cache keys protocol for key generation.

Defines the port for cache key construction. Infrastructure layer implements this protocol to provide consistent key patterns.

Architecture
  • Domain layer protocol (port)
  • Infrastructure adapter: src/infrastructure/cache/cache_keys.py
  • Used by handlers for cache key generation
Reference
  • docs/architecture/hexagonal.md
  • docs/architecture/caching-architecture.md

Classes

CacheKeysProtocol

Bases: Protocol

Protocol for generating cache keys.

Abstracts cache key construction used by application layer handlers. Infrastructure layer provides concrete implementation.

All keys follow hierarchical structure: {prefix}:{domain}:{resource}:{id}

Example

class RefreshAccessTokenHandler: def init( self, cache_keys: CacheKeysProtocol | None = None, ... ) -> None: self._cache_keys = cache_keys

Source code in src/domain/protocols/cache_keys_protocol.py 
 21
 22
 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
class CacheKeysProtocol(Protocol):
    """Protocol for generating cache keys.

    Abstracts cache key construction used by application layer handlers.
    Infrastructure layer provides concrete implementation.

    All keys follow hierarchical structure: {prefix}:{domain}:{resource}:{id}

    Example:
        class RefreshAccessTokenHandler:
            def __init__(
                self,
                cache_keys: CacheKeysProtocol | None = None,
                ...
            ) -> None:
                self._cache_keys = cache_keys
    """

    @property
    def prefix(self) -> str:
        """Cache key prefix (typically \"dashtam\")."""
        ...

    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.
        """
        ...

    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.
        """
        ...

    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.
        """
        ...

    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.
        """
        ...

    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.
        """
        ...

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

        Pattern: {prefix}:security:global_version

        Returns:
            Cache key string.

        Note:
            Used for token breach rotation (F1.3b).
        """
        ...

    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.

        Note:
            Used for token breach rotation (F1.3b).
        """
        ...

    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\").
        """
        ...
Attributes
prefix property 
prefix: str

Cache key prefix (typically "dashtam").

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.
Source code in src/domain/protocols/cache_keys_protocol.py 
44
45
46
47
48
49
50
51
52
53
54
55
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.
    """
    ...
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.
Source code in src/domain/protocols/cache_keys_protocol.py 
57
58
59
60
61
62
63
64
65
66
67
68
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.
    """
    ...
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.
Source code in src/domain/protocols/cache_keys_protocol.py 
70
71
72
73
74
75
76
77
78
79
80
81
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.
    """
    ...
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.
Source code in src/domain/protocols/cache_keys_protocol.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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.
    """
    ...
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.
Source code in src/domain/protocols/cache_keys_protocol.py 
103
104
105
106
107
108
109
110
111
112
113
114
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.
    """
    ...
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.
Note

Used for token breach rotation (F1.3b).

Source code in src/domain/protocols/cache_keys_protocol.py 
116
117
118
119
120
121
122
123
124
125
126
127
def security_global_version(self) -> str:
    """Security global token version cache key.

    Pattern: {prefix}:security:global_version

    Returns:
        Cache key string.

    Note:
        Used for token breach rotation (F1.3b).
    """
    ...
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.
Note

Used for token breach rotation (F1.3b).

Source code in src/domain/protocols/cache_keys_protocol.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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.

    Note:
        Used for token breach rotation (F1.3b).
    """
    ...
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").
Source code in src/domain/protocols/cache_keys_protocol.py 
145
146
147
148
149
150
151
152
153
154
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\").
    """
    ...