Skip to content

domain.protocols.session_cache_protocol

src.domain.protocols.session_cache_protocol

Session cache protocol for fast session lookups.

This module defines the port (interface) for session caching. Infrastructure layer implements with Redis for <5ms lookups.

Reference
  • docs/architecture/session-management-architecture.md

Classes

SessionCache

Bases: Protocol

Session cache protocol (port) for fast lookups.

Provides <5ms session lookups via Redis caching. Acts as write-through cache: writes go to both cache and database.

Cache Strategy
  • Session data cached on creation
  • Cache invalidated on revocation/update
  • TTL matches session expiration
  • Database is source of truth
Key Patterns
  • session:{session_id} -> SessionData
  • user:{user_id}:sessions -> Set of session IDs
  • session:{session_id}:validation -> Quick validation data
Example

class RedisSessionCache: ... async def get(self, session_id: UUID) -> SessionData | None: ... # Look up in Redis ... ... ... async def set(self, session_data: SessionData) -> None: ... # Store in Redis with TTL ... ...

Source code in src/domain/protocols/session_cache_protocol.py 
 16
 17
 18
 19
 20
 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
155
156
157
class SessionCache(Protocol):
    """Session cache protocol (port) for fast lookups.

    Provides <5ms session lookups via Redis caching.
    Acts as write-through cache: writes go to both cache and database.

    Cache Strategy:
        - Session data cached on creation
        - Cache invalidated on revocation/update
        - TTL matches session expiration
        - Database is source of truth

    Key Patterns:
        - session:{session_id} -> SessionData
        - user:{user_id}:sessions -> Set of session IDs
        - session:{session_id}:validation -> Quick validation data

    Example:
        >>> class RedisSessionCache:
        ...     async def get(self, session_id: UUID) -> SessionData | None:
        ...         # Look up in Redis
        ...         ...
        ...     async def set(self, session_data: SessionData) -> None:
        ...         # Store in Redis with TTL
        ...         ...
    """

    async def get(self, session_id: UUID) -> SessionData | None:
        """Get session data from cache.

        Args:
            session_id: Session identifier.

        Returns:
            SessionData if cached, None otherwise.
        """
        ...

    async def set(
        self,
        session_data: SessionData,
        *,
        ttl_seconds: int | None = None,
    ) -> None:
        """Store session data in cache.

        Args:
            session_data: Session data to cache.
            ttl_seconds: Cache TTL in seconds. If None, calculates from
                session expires_at. Defaults to 30 days if no expiry.
        """
        ...

    async def delete(self, session_id: UUID) -> bool:
        """Remove session from cache.

        Called when session is revoked or expired.

        Args:
            session_id: Session identifier.

        Returns:
            True if deleted, False if not found.
        """
        ...

    async def delete_all_for_user(self, user_id: UUID) -> int:
        """Remove all sessions for a user from cache.

        Called when all sessions are revoked (e.g., password change).

        Args:
            user_id: User identifier.

        Returns:
            Number of sessions removed from cache.
        """
        ...

    async def exists(self, session_id: UUID) -> bool:
        """Check if session exists in cache (quick validation).

        Faster than full get() when only existence check needed.

        Args:
            session_id: Session identifier.

        Returns:
            True if session exists in cache, False otherwise.
        """
        ...

    async def get_user_session_ids(self, user_id: UUID) -> list[UUID]:
        """Get all session IDs for a user from cache.

        Args:
            user_id: User identifier.

        Returns:
            List of session IDs, empty if none cached.
        """
        ...

    async def add_user_session(self, user_id: UUID, session_id: UUID) -> None:
        """Add session ID to user's session set.

        Called when new session created. Maintains reverse index.

        Args:
            user_id: User identifier.
            session_id: Session identifier.
        """
        ...

    async def remove_user_session(self, user_id: UUID, session_id: UUID) -> None:
        """Remove session ID from user's session set.

        Called when session revoked/deleted.

        Args:
            user_id: User identifier.
            session_id: Session identifier.
        """
        ...

    async def update_last_activity(
        self,
        session_id: UUID,
        ip_address: str | None = None,
    ) -> bool:
        """Update session's last activity in cache.

        Lightweight update for activity tracking.

        Args:
            session_id: Session identifier.
            ip_address: Current IP address (optional).

        Returns:
            True if updated, False if session not in cache.
        """
        ...
Functions
get async 
get(session_id: UUID) -> SessionData | None

Get session data from cache.

Parameters:

Name Type Description Default
session_id UUID

Session identifier.

 required

Returns:

Type Description
SessionData | None

SessionData if cached, None otherwise.
Source code in src/domain/protocols/session_cache_protocol.py 
43
44
45
46
47
48
49
50
51
52
async def get(self, session_id: UUID) -> SessionData | None:
    """Get session data from cache.

    Args:
        session_id: Session identifier.

    Returns:
        SessionData if cached, None otherwise.
    """
    ...
set async 
set(
    session_data: SessionData,
    *,
    ttl_seconds: int | None = None
) -> None

Store session data in cache.

Parameters:

Name Type Description Default
session_data SessionData

Session data to cache.

 required
ttl_seconds int | None

Cache TTL in seconds. If None, calculates from session expires_at. Defaults to 30 days if no expiry.

 None
Source code in src/domain/protocols/session_cache_protocol.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def set(
    self,
    session_data: SessionData,
    *,
    ttl_seconds: int | None = None,
) -> None:
    """Store session data in cache.

    Args:
        session_data: Session data to cache.
        ttl_seconds: Cache TTL in seconds. If None, calculates from
            session expires_at. Defaults to 30 days if no expiry.
    """
    ...
delete async 
delete(session_id: UUID) -> bool

Remove session from cache.

Called when session is revoked or expired.

Parameters:

Name Type Description Default
session_id UUID

Session identifier.

 required

Returns:

Type Description
bool

True if deleted, False if not found.
Source code in src/domain/protocols/session_cache_protocol.py 
69
70
71
72
73
74
75
76
77
78
79
80
async def delete(self, session_id: UUID) -> bool:
    """Remove session from cache.

    Called when session is revoked or expired.

    Args:
        session_id: Session identifier.

    Returns:
        True if deleted, False if not found.
    """
    ...
delete_all_for_user async 
delete_all_for_user(user_id: UUID) -> int

Remove all sessions for a user from cache.

Called when all sessions are revoked (e.g., password change).

Parameters:

Name Type Description Default
user_id UUID

User identifier.

 required

Returns:

Type Description
int

Number of sessions removed from cache.
Source code in src/domain/protocols/session_cache_protocol.py 
82
83
84
85
86
87
88
89
90
91
92
93
async def delete_all_for_user(self, user_id: UUID) -> int:
    """Remove all sessions for a user from cache.

    Called when all sessions are revoked (e.g., password change).

    Args:
        user_id: User identifier.

    Returns:
        Number of sessions removed from cache.
    """
    ...
exists async 
exists(session_id: UUID) -> bool

Check if session exists in cache (quick validation).

Faster than full get() when only existence check needed.

Parameters:

Name Type Description Default
session_id UUID

Session identifier.

 required

Returns:

Type Description
bool

True if session exists in cache, False otherwise.
Source code in src/domain/protocols/session_cache_protocol.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def exists(self, session_id: UUID) -> bool:
    """Check if session exists in cache (quick validation).

    Faster than full get() when only existence check needed.

    Args:
        session_id: Session identifier.

    Returns:
        True if session exists in cache, False otherwise.
    """
    ...
get_user_session_ids async 
get_user_session_ids(user_id: UUID) -> list[UUID]

Get all session IDs for a user from cache.

Parameters:

Name Type Description Default
user_id UUID

User identifier.

 required

Returns:

Type Description
list[UUID]

List of session IDs, empty if none cached.
Source code in src/domain/protocols/session_cache_protocol.py 
108
109
110
111
112
113
114
115
116
117
async def get_user_session_ids(self, user_id: UUID) -> list[UUID]:
    """Get all session IDs for a user from cache.

    Args:
        user_id: User identifier.

    Returns:
        List of session IDs, empty if none cached.
    """
    ...
add_user_session async 
add_user_session(user_id: UUID, session_id: UUID) -> None

Add session ID to user's session set.

Called when new session created. Maintains reverse index.

Parameters:

Name Type Description Default
user_id UUID

User identifier.

 required
session_id UUID

Session identifier.

 required
Source code in src/domain/protocols/session_cache_protocol.py 
119
120
121
122
123
124
125
126
127
128
async def add_user_session(self, user_id: UUID, session_id: UUID) -> None:
    """Add session ID to user's session set.

    Called when new session created. Maintains reverse index.

    Args:
        user_id: User identifier.
        session_id: Session identifier.
    """
    ...
remove_user_session async 
remove_user_session(
    user_id: UUID, session_id: UUID
) -> None

Remove session ID from user's session set.

Called when session revoked/deleted.

Parameters:

Name Type Description Default
user_id UUID

User identifier.

 required
session_id UUID

Session identifier.

 required
Source code in src/domain/protocols/session_cache_protocol.py 
130
131
132
133
134
135
136
137
138
139
async def remove_user_session(self, user_id: UUID, session_id: UUID) -> None:
    """Remove session ID from user's session set.

    Called when session revoked/deleted.

    Args:
        user_id: User identifier.
        session_id: Session identifier.
    """
    ...
update_last_activity async 
update_last_activity(
    session_id: UUID, ip_address: str | None = None
) -> bool

Update session's last activity in cache.

Lightweight update for activity tracking.

Parameters:

Name Type Description Default
session_id UUID

Session identifier.

 required
ip_address str | None

Current IP address (optional).

 None

Returns:

Type Description
bool

True if updated, False if session not in cache.
Source code in src/domain/protocols/session_cache_protocol.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def update_last_activity(
    self,
    session_id: UUID,
    ip_address: str | None = None,
) -> bool:
    """Update session's last activity in cache.

    Lightweight update for activity tracking.

    Args:
        session_id: Session identifier.
        ip_address: Current IP address (optional).

    Returns:
        True if updated, False if session not in cache.
    """
    ...