Skip to content

application.commands.handlers.trigger_user_rotation_handler

src.application.commands.handlers.trigger_user_rotation_handler

Handler for TriggerUserTokenRotation command.

Flow: 1. Emit UserTokenRotationAttempted event 2. Get user from database 3. Verify user exists 4. Increment user.min_token_version 5. Save user 6. Emit UserTokenRotationSucceeded event 7. Return Success(UserRotationResult)

On failure: - Emit UserTokenRotationFailed event - Return Failure(error)

Classes

RotationError

User rotation error reasons.

Source code in src/application/commands/handlers/trigger_user_rotation_handler.py 
32
33
34
35
class RotationError:
    """User rotation error reasons."""

    USER_NOT_FOUND = "user_not_found"

TriggerUserTokenRotationHandler

Handler for per-user token rotation command.

Increments the user's minimum token version, which causes only that user's existing refresh tokens to fail validation.

Use cases: - Password change (automatic integration) - "Log out everywhere" user action - Admin action on suspicious accounts

Source code in src/application/commands/handlers/trigger_user_rotation_handler.py 
 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
class TriggerUserTokenRotationHandler:
    """Handler for per-user token rotation command.

    Increments the user's minimum token version, which causes only
    that user's existing refresh tokens to fail validation.

    Use cases:
    - Password change (automatic integration)
    - "Log out everywhere" user action
    - Admin action on suspicious accounts
    """

    def __init__(
        self,
        user_repo: UserRepository,
        event_bus: EventBusProtocol,
    ) -> None:
        """Initialize handler with dependencies.

        Args:
            user_repo: User repository for persistence.
            event_bus: Event bus for publishing domain events.
        """
        self._user_repo = user_repo
        self._event_bus = event_bus

    async def handle(
        self,
        cmd: TriggerUserTokenRotation,
    ) -> Result[UserRotationResult, str]:
        """Handle per-user token rotation command.

        Args:
            cmd: TriggerUserTokenRotation command.

        Returns:
            Success(UserRotationResult) on success.
            Failure(error) on failure.
        """
        # Step 1: Emit ATTEMPTED event
        await self._event_bus.publish(
            UserTokenRotationAttempted(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                triggered_by=cmd.triggered_by,
                reason=cmd.reason,
            )
        )

        # Step 2: Get user from database
        user = await self._user_repo.find_by_id(cmd.user_id)

        # Step 3: Verify user exists
        if user is None:
            await self._event_bus.publish(
                UserTokenRotationFailed(
                    event_id=uuid7(),
                    occurred_at=datetime.now(UTC),
                    user_id=cmd.user_id,
                    triggered_by=cmd.triggered_by,
                    reason=cmd.reason,
                    failure_reason=RotationError.USER_NOT_FOUND,
                )
            )
            return Failure(error=RotationError.USER_NOT_FOUND)

        # Step 4: Increment user.min_token_version
        previous_version = user.min_token_version
        user.min_token_version = previous_version + 1
        user.updated_at = datetime.now(UTC)

        # Step 5: Save user
        await self._user_repo.update(user)

        # Step 6: Emit SUCCEEDED event
        await self._event_bus.publish(
            UserTokenRotationSucceeded(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                triggered_by=cmd.triggered_by,
                previous_version=previous_version,
                new_version=user.min_token_version,
                reason=cmd.reason,
            )
        )

        # Step 7: Return success
        return Success(
            value=UserRotationResult(
                user_id=cmd.user_id,
                previous_version=previous_version,
                new_version=user.min_token_version,
            )
        )
Functions
__init__ 
__init__(
    user_repo: UserRepository, event_bus: EventBusProtocol
) -> None

Parameters:

Name Type Description Default
user_repo UserRepository

User repository for persistence.

 required
event_bus EventBusProtocol

Event bus for publishing domain events.

 required
Source code in src/application/commands/handlers/trigger_user_rotation_handler.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    user_repo: UserRepository,
    event_bus: EventBusProtocol,
) -> None:
    """Initialize handler with dependencies.

    Args:
        user_repo: User repository for persistence.
        event_bus: Event bus for publishing domain events.
    """
    self._user_repo = user_repo
    self._event_bus = event_bus
handle async 
handle(
    cmd: TriggerUserTokenRotation,
) -> Result[UserRotationResult, str]

Handle per-user token rotation command.

Parameters:

Name Type Description Default
cmd TriggerUserTokenRotation

TriggerUserTokenRotation command.

 required

Returns:

Type Description
Result[UserRotationResult, str]

Success(UserRotationResult) on success.
Result[UserRotationResult, str]

Failure(error) on failure.
Source code in src/application/commands/handlers/trigger_user_rotation_handler.py 
 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
async def handle(
    self,
    cmd: TriggerUserTokenRotation,
) -> Result[UserRotationResult, str]:
    """Handle per-user token rotation command.

    Args:
        cmd: TriggerUserTokenRotation command.

    Returns:
        Success(UserRotationResult) on success.
        Failure(error) on failure.
    """
    # Step 1: Emit ATTEMPTED event
    await self._event_bus.publish(
        UserTokenRotationAttempted(
            event_id=uuid7(),
            occurred_at=datetime.now(UTC),
            user_id=cmd.user_id,
            triggered_by=cmd.triggered_by,
            reason=cmd.reason,
        )
    )

    # Step 2: Get user from database
    user = await self._user_repo.find_by_id(cmd.user_id)

    # Step 3: Verify user exists
    if user is None:
        await self._event_bus.publish(
            UserTokenRotationFailed(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                triggered_by=cmd.triggered_by,
                reason=cmd.reason,
                failure_reason=RotationError.USER_NOT_FOUND,
            )
        )
        return Failure(error=RotationError.USER_NOT_FOUND)

    # Step 4: Increment user.min_token_version
    previous_version = user.min_token_version
    user.min_token_version = previous_version + 1
    user.updated_at = datetime.now(UTC)

    # Step 5: Save user
    await self._user_repo.update(user)

    # Step 6: Emit SUCCEEDED event
    await self._event_bus.publish(
        UserTokenRotationSucceeded(
            event_id=uuid7(),
            occurred_at=datetime.now(UTC),
            user_id=cmd.user_id,
            triggered_by=cmd.triggered_by,
            previous_version=previous_version,
            new_version=user.min_token_version,
            reason=cmd.reason,
        )
    )

    # Step 7: Return success
    return Success(
        value=UserRotationResult(
            user_id=cmd.user_id,
            previous_version=previous_version,
            new_version=user.min_token_version,
        )
    )