Skip to content

application.queries.handlers.get_session_handler

src.application.queries.handlers.get_session_handler

Get session query handler.

Retrieves a single session by ID with authorization check. Uses cache-first strategy for performance.

Classes

GetSessionError

Get session error reasons.

Source code in src/application/queries/handlers/get_session_handler.py 
17
18
19
20
21
class GetSessionError:
    """Get session error reasons."""

    SESSION_NOT_FOUND = "session_not_found"
    NOT_OWNER = "not_session_owner"

SessionResult dataclass

Session query result.

Source code in src/application/queries/handlers/get_session_handler.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass
class SessionResult:
    """Session query result."""

    id: UUID
    user_id: UUID
    device_info: str | None
    ip_address: str | None
    location: str | None
    created_at: datetime | None
    last_activity_at: datetime | None
    expires_at: datetime | None
    is_revoked: bool
    is_current: bool = False

GetSessionHandler

Handler for getting a single session.

Uses cache-first strategy: 1. Try cache 2. Fall back to database 3. Populate cache on miss

Source code in src/application/queries/handlers/get_session_handler.py 
 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
class GetSessionHandler:
    """Handler for getting a single session.

    Uses cache-first strategy:
    1. Try cache
    2. Fall back to database
    3. Populate cache on miss
    """

    def __init__(
        self,
        session_repo: SessionRepository,
        session_cache: SessionCache,
    ) -> None:
        """Initialize handler with dependencies.

        Args:
            session_repo: Session repository for persistence.
            session_cache: Session cache for fast lookups.
        """
        self._session_repo = session_repo
        self._session_cache = session_cache

    async def handle(self, query: GetSession) -> Result[SessionResult, str]:
        """Handle get session query.

        Args:
            query: GetSession query with session_id and user_id.

        Returns:
            Success(SessionResult) with session data.
            Failure(error_message) if not found or not owner.
        """
        # Step 1: Try cache first
        session = await self._session_cache.get(query.session_id)

        # Step 2: Fall back to database
        if session is None:
            session = await self._session_repo.find_by_id(query.session_id)

            # Populate cache on miss (if found and not revoked)
            if session is not None and not session.is_revoked:
                await self._session_cache.set(session)

        # Step 3: Check found
        if session is None:
            return Failure(error=GetSessionError.SESSION_NOT_FOUND)

        # Step 4: Check ownership
        if session.user_id != query.user_id:
            return Failure(error=GetSessionError.NOT_OWNER)

        # Step 5: Return result
        return Success(
            value=SessionResult(
                id=session.id,
                user_id=session.user_id,
                device_info=session.device_info,
                ip_address=session.ip_address,
                location=session.location,
                created_at=session.created_at,
                last_activity_at=session.last_activity_at,
                expires_at=session.expires_at,
                is_revoked=session.is_revoked,
                is_current=False,  # Caller sets this
            )
        )
Functions
__init__ 
__init__(
    session_repo: SessionRepository,
    session_cache: SessionCache,
) -> None

Parameters:

Name Type Description Default
session_repo SessionRepository

Session repository for persistence.

 required
session_cache SessionCache

Session cache for fast lookups.

 required
Source code in src/application/queries/handlers/get_session_handler.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    session_repo: SessionRepository,
    session_cache: SessionCache,
) -> None:
    """Initialize handler with dependencies.

    Args:
        session_repo: Session repository for persistence.
        session_cache: Session cache for fast lookups.
    """
    self._session_repo = session_repo
    self._session_cache = session_cache
handle async 
handle(query: GetSession) -> Result[SessionResult, str]

Handle get session query.

Parameters:

Name Type Description Default
query GetSession

GetSession query with session_id and user_id.

 required

Returns:

Type Description
Result[SessionResult, str]

Success(SessionResult) with session data.
Result[SessionResult, str]

Failure(error_message) if not found or not owner.
Source code in src/application/queries/handlers/get_session_handler.py 
 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
async def handle(self, query: GetSession) -> Result[SessionResult, str]:
    """Handle get session query.

    Args:
        query: GetSession query with session_id and user_id.

    Returns:
        Success(SessionResult) with session data.
        Failure(error_message) if not found or not owner.
    """
    # Step 1: Try cache first
    session = await self._session_cache.get(query.session_id)

    # Step 2: Fall back to database
    if session is None:
        session = await self._session_repo.find_by_id(query.session_id)

        # Populate cache on miss (if found and not revoked)
        if session is not None and not session.is_revoked:
            await self._session_cache.set(session)

    # Step 3: Check found
    if session is None:
        return Failure(error=GetSessionError.SESSION_NOT_FOUND)

    # Step 4: Check ownership
    if session.user_id != query.user_id:
        return Failure(error=GetSessionError.NOT_OWNER)

    # Step 5: Return result
    return Success(
        value=SessionResult(
            id=session.id,
            user_id=session.user_id,
            device_info=session.device_info,
            ip_address=session.ip_address,
            location=session.location,
            created_at=session.created_at,
            last_activity_at=session.last_activity_at,
            expires_at=session.expires_at,
            is_revoked=session.is_revoked,
            is_current=False,  # Caller sets this
        )
    )