Skip to content

application.commands.session_commands

src.application.commands.session_commands

Session management commands (CQRS write operations).

Commands represent user intent to change session state. All commands are immutable (frozen=True) and use keyword-only arguments (kw_only=True).

Pattern: - Commands are data containers (no logic) - Handlers execute business logic - Commands don't return values (handlers return Result types)

Classes

CreateSession dataclass

Create a new session for a user.

Called during login to create session with device/location metadata. Enforces session limits per user tier.

Attributes:

Name Type Description
user_id UUID

User identifier.
ip_address str | None

Client IP address (for geolocation, audit).
user_agent str | None

Client user agent (for device parsing).
expires_at datetime | None

When session expires (matches refresh token).
refresh_token_id UUID | None

Associated refresh token ID (optional, set after token created).
Example

command = CreateSession( ... user_id=UUID("123e4567-e89b-12d3-a456-426614174000"), ... ip_address="192.168.1.1", ... user_agent="Mozilla/5.0...", ... expires_at=datetime.now(UTC) + timedelta(days=30), ... ) result = await handler.handle(command)

Source code in src/application/commands/session_commands.py 
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
@dataclass(frozen=True, kw_only=True)
class CreateSession:
    """Create a new session for a user.

    Called during login to create session with device/location metadata.
    Enforces session limits per user tier.

    Attributes:
        user_id: User identifier.
        ip_address: Client IP address (for geolocation, audit).
        user_agent: Client user agent (for device parsing).
        expires_at: When session expires (matches refresh token).
        refresh_token_id: Associated refresh token ID (optional, set after token created).

    Example:
        >>> command = CreateSession(
        ...     user_id=UUID("123e4567-e89b-12d3-a456-426614174000"),
        ...     ip_address="192.168.1.1",
        ...     user_agent="Mozilla/5.0...",
        ...     expires_at=datetime.now(UTC) + timedelta(days=30),
        ... )
        >>> result = await handler.handle(command)
    """

    user_id: UUID
    ip_address: str | None = None
    user_agent: str | None = None
    expires_at: datetime | None = None
    refresh_token_id: UUID | None = None

RevokeSession dataclass

Revoke a specific session.

Called on logout or when user manually revokes a session. Soft-deletes session (marks as revoked, keeps for audit).

Attributes:

Name Type Description
session_id UUID

Session identifier to revoke.
user_id UUID

User identifier (for authorization check).
reason str

Revocation reason for audit (logout, manual, suspicious, etc.).
Example

command = RevokeSession( ... session_id=UUID("abc123..."), ... user_id=UUID("123e4567..."), ... reason="logout", ... ) result = await handler.handle(command)

Source code in src/application/commands/session_commands.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass(frozen=True, kw_only=True)
class RevokeSession:
    """Revoke a specific session.

    Called on logout or when user manually revokes a session.
    Soft-deletes session (marks as revoked, keeps for audit).

    Attributes:
        session_id: Session identifier to revoke.
        user_id: User identifier (for authorization check).
        reason: Revocation reason for audit (logout, manual, suspicious, etc.).

    Example:
        >>> command = RevokeSession(
        ...     session_id=UUID("abc123..."),
        ...     user_id=UUID("123e4567..."),
        ...     reason="logout",
        ... )
        >>> result = await handler.handle(command)
    """

    session_id: UUID
    user_id: UUID
    reason: str = "logout"

RevokeAllUserSessions dataclass

Revoke all sessions for a user.

Called on password change, security event, or user-initiated "logout everywhere". Can optionally exclude current session.

Attributes:

Name Type Description
user_id UUID

User identifier.
reason str

Revocation reason for audit.
except_session_id UUID | None

Session ID to exclude (current session).
Example

command = RevokeAllUserSessions( ... user_id=UUID("123e4567..."), ... reason="password_change", ... except_session_id=current_session_id, ... ) result = await handler.handle(command)

Source code in src/application/commands/session_commands.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@dataclass(frozen=True, kw_only=True)
class RevokeAllUserSessions:
    """Revoke all sessions for a user.

    Called on password change, security event, or user-initiated "logout everywhere".
    Can optionally exclude current session.

    Attributes:
        user_id: User identifier.
        reason: Revocation reason for audit.
        except_session_id: Session ID to exclude (current session).

    Example:
        >>> command = RevokeAllUserSessions(
        ...     user_id=UUID("123e4567..."),
        ...     reason="password_change",
        ...     except_session_id=current_session_id,
        ... )
        >>> result = await handler.handle(command)
    """

    user_id: UUID
    reason: str
    except_session_id: UUID | None = None

UpdateSessionActivity dataclass

Update session's last activity timestamp.

Called on each authenticated request to track activity. Lightweight operation - updates cache and database.

Attributes:

Name Type Description
session_id UUID

Session identifier.
ip_address str | None

Current client IP (for detecting IP changes).
Example

command = UpdateSessionActivity( ... session_id=UUID("abc123..."), ... ip_address="192.168.1.100", ... ) await handler.handle(command)

Source code in src/application/commands/session_commands.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@dataclass(frozen=True, kw_only=True)
class UpdateSessionActivity:
    """Update session's last activity timestamp.

    Called on each authenticated request to track activity.
    Lightweight operation - updates cache and database.

    Attributes:
        session_id: Session identifier.
        ip_address: Current client IP (for detecting IP changes).

    Example:
        >>> command = UpdateSessionActivity(
        ...     session_id=UUID("abc123..."),
        ...     ip_address="192.168.1.100",
        ... )
        >>> await handler.handle(command)
    """

    session_id: UUID
    ip_address: str | None = None

RecordProviderAccess dataclass

Record provider access in session.

Called when user accesses a financial provider (Schwab, etc.). Tracks which providers accessed per session for audit.

Attributes:

Name Type Description
session_id UUID

Session identifier.
provider_name str

Name of provider accessed.
Example

command = RecordProviderAccess( ... session_id=UUID("abc123..."), ... provider_name="schwab", ... ) await handler.handle(command)

Source code in src/application/commands/session_commands.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@dataclass(frozen=True, kw_only=True)
class RecordProviderAccess:
    """Record provider access in session.

    Called when user accesses a financial provider (Schwab, etc.).
    Tracks which providers accessed per session for audit.

    Attributes:
        session_id: Session identifier.
        provider_name: Name of provider accessed.

    Example:
        >>> command = RecordProviderAccess(
        ...     session_id=UUID("abc123..."),
        ...     provider_name="schwab",
        ... )
        >>> await handler.handle(command)
    """

    session_id: UUID
    provider_name: str

LinkRefreshTokenToSession dataclass

Link a refresh token to a session.

Called after refresh token is created to establish the relationship.

Attributes:

Name Type Description
session_id UUID

Session identifier.
refresh_token_id UUID

Refresh token identifier.
Example

command = LinkRefreshTokenToSession( ... session_id=UUID("abc123..."), ... refresh_token_id=UUID("def456..."), ... ) await handler.handle(command)

Source code in src/application/commands/session_commands.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@dataclass(frozen=True, kw_only=True)
class LinkRefreshTokenToSession:
    """Link a refresh token to a session.

    Called after refresh token is created to establish the relationship.

    Attributes:
        session_id: Session identifier.
        refresh_token_id: Refresh token identifier.

    Example:
        >>> command = LinkRefreshTokenToSession(
        ...     session_id=UUID("abc123..."),
        ...     refresh_token_id=UUID("def456..."),
        ... )
        >>> await handler.handle(command)
    """

    session_id: UUID
    refresh_token_id: UUID