presentation.routers.api.v1.sessions

src.presentation.routers.api.v1.sessions

Sessions resource handlers.

Handler functions for session management endpoints. Routes are registered via ROUTE_REGISTRY in routes/registry.py.

Handlers

create_session - Create session (login) delete_current_session - Delete current session (logout) list_sessions - List user sessions get_session - Get session details revoke_session - Revoke specific session revoke_all_sessions - Revoke all sessions (except current)

Classes

Functions

create_session async

create_session(
    request: Request,
    data: SessionCreateRequest,
    auth_handler: AuthenticateUserHandler = Depends(
        handler_factory(AuthenticateUserHandler)
    ),
    session_handler: CreateSessionHandler = Depends(
        handler_factory(CreateSessionHandler)
    ),
    token_handler: GenerateAuthTokensHandler = Depends(
        handler_factory(GenerateAuthTokensHandler)
    ),
) -> SessionCreateResponse | JSONResponse

Create a new session (login).

POST /api/v1/sessions → 201 Created

Orchestrates 3 handlers (CQRS pattern): 1. AuthenticateUser - Verify credentials 2. CreateSession - Create session with device/location 3. GenerateAuthTokens - Generate JWT + refresh token

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
data	SessionCreateRequest	Session creation request (email, password).	required
auth_handler	AuthenticateUserHandler	Authentication handler (injected).	Depends(handler_factory(AuthenticateUserHandler))
session_handler	CreateSessionHandler	Session creation handler (injected).	Depends(handler_factory(CreateSessionHandler))
token_handler	GenerateAuthTokensHandler	Token generation handler (injected).	Depends(handler_factory(GenerateAuthTokensHandler))

Returns:

Type	Description
SessionCreateResponse | JSONResponse	SessionCreateResponse on success (201 Created).
SessionCreateResponse | JSONResponse	JSONResponse with error on failure (400/401/403).

Source code in src/presentation/routers/api/v1/sessions.py

async def create_session(
    request: Request,
    data: SessionCreateRequest,
    auth_handler: AuthenticateUserHandler = Depends(
        handler_factory(AuthenticateUserHandler)
    ),
    session_handler: CreateSessionHandler = Depends(
        handler_factory(CreateSessionHandler)
    ),
    token_handler: GenerateAuthTokensHandler = Depends(
        handler_factory(GenerateAuthTokensHandler)
    ),
) -> SessionCreateResponse | JSONResponse:
    """Create a new session (login).

    POST /api/v1/sessions → 201 Created

    Orchestrates 3 handlers (CQRS pattern):
    1. AuthenticateUser - Verify credentials
    2. CreateSession - Create session with device/location
    3. GenerateAuthTokens - Generate JWT + refresh token

    Args:
        request: FastAPI request object.
        data: Session creation request (email, password).
        auth_handler: Authentication handler (injected).
        session_handler: Session creation handler (injected).
        token_handler: Token generation handler (injected).

    Returns:
        SessionCreateResponse on success (201 Created).
        JSONResponse with error on failure (400/401/403).
    """
    # Extract client info for session
    ip_address = request.client.host if request.client else None
    user_agent = request.headers.get("user-agent")

    # Step 1: Authenticate user credentials
    auth_command = AuthenticateUser(
        email=data.email,
        password=data.password,
    )
    auth_result = await auth_handler.handle(auth_command, request)

    match auth_result:
        case Failure(error=error):
            return _create_auth_error_response(request, error)
        case Success(value=authenticated_user):
            pass  # Continue to step 2

    # Step 2: Create session
    session_command = CreateSession(
        user_id=authenticated_user.user_id,
        ip_address=ip_address,
        user_agent=user_agent,
    )
    session_result = await session_handler.handle(session_command)

    match session_result:
        case Failure(error=error):
            return _create_auth_error_response(request, error)
        case Success(value=session_response):
            pass  # Continue to step 3

    # Step 3: Generate tokens
    token_command = GenerateAuthTokens(
        user_id=authenticated_user.user_id,
        email=authenticated_user.email,
        roles=authenticated_user.roles,
        session_id=session_response.session_id,
    )
    token_result = await token_handler.handle(token_command)

    match token_result:
        case Failure(error=error):
            return _create_auth_error_response(request, error)
        case Success(value=tokens):
            return SessionCreateResponse(
                access_token=tokens.access_token,
                refresh_token=tokens.refresh_token,
                token_type=tokens.token_type,
                expires_in=tokens.expires_in,
            )

delete_current_session async

delete_current_session(
    request: Request,
    data: SessionDeleteRequest,
    authorization: Annotated[str | None, Header()] = None,
    handler: LogoutUserHandler = Depends(
        handler_factory(LogoutUserHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> Response | JSONResponse

Delete current session (logout).

DELETE /api/v1/sessions/current → 204 No Content

Revokes the refresh token to prevent new access tokens. The current access token remains valid until expiration (15 min).

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
data	SessionDeleteRequest	Session delete request (refresh_token).	required
authorization	Annotated[str | None, Header()]	JWT access token from Authorization header.	None
handler	LogoutUserHandler	Logout handler (injected).	Depends(handler_factory(LogoutUserHandler))

Returns:

Type	Description
Response | JSONResponse	204 No Content on success.
Response | JSONResponse	JSONResponse with error on failure (401).

Source code in src/presentation/routers/api/v1/sessions.py

async def delete_current_session(
    request: Request,
    data: SessionDeleteRequest,
    authorization: Annotated[str | None, Header()] = None,
    handler: LogoutUserHandler = Depends(handler_factory(LogoutUserHandler)),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> Response | JSONResponse:
    """Delete current session (logout).

    DELETE /api/v1/sessions/current → 204 No Content

    Revokes the refresh token to prevent new access tokens.
    The current access token remains valid until expiration (15 min).

    Args:
        request: FastAPI request object.
        data: Session delete request (refresh_token).
        authorization: JWT access token from Authorization header.
        handler: Logout handler (injected).

    Returns:
        204 No Content on success.
        JSONResponse with error on failure (401).
    """
    # Extract user_id from JWT token (if provided)
    user_id = await _extract_user_id_from_token(authorization, cache, db_session)

    if user_id is None:
        app_error = ApplicationError(
            code=ApplicationErrorCode.UNAUTHORIZED,
            message="Valid authorization token required",
        )
        return ErrorResponseBuilder.from_application_error(
            error=app_error,
            request=request,
            trace_id=get_trace_id() or "",
        )

    # Create command
    command = LogoutUser(
        user_id=user_id,
        refresh_token=data.refresh_token,
    )

    # Execute handler (always returns success for security)
    await handler.handle(command, request)

    # Return 204 No Content
    return Response(status_code=status.HTTP_204_NO_CONTENT)

list_sessions async

list_sessions(
    request: Request,
    authorization: Annotated[str | None, Header()] = None,
    active_only: bool = Query(
        default=True,
        description="Only return active sessions",
    ),
    handler: ListSessionsHandler = Depends(
        handler_factory(ListSessionsHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionListResponse | JSONResponse

List all sessions for the current user.

GET /api/v1/sessions → 200 OK

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
authorization	Annotated[str | None, Header()]	JWT access token from Authorization header.	None
active_only	bool	Filter to active sessions only.	Query(default=True, description='Only return active sessions')
handler	ListSessionsHandler	List sessions handler (injected).	Depends(handler_factory(ListSessionsHandler))

Returns:

Type	Description
SessionListResponse | JSONResponse	SessionListResponse with list of sessions.
SessionListResponse | JSONResponse	JSONResponse with error on failure (401).

Source code in src/presentation/routers/api/v1/sessions.py

async def list_sessions(
    request: Request,
    authorization: Annotated[str | None, Header()] = None,
    active_only: bool = Query(default=True, description="Only return active sessions"),
    handler: ListSessionsHandler = Depends(handler_factory(ListSessionsHandler)),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionListResponse | JSONResponse:
    """List all sessions for the current user.

    GET /api/v1/sessions → 200 OK

    Args:
        request: FastAPI request object.
        authorization: JWT access token from Authorization header.
        active_only: Filter to active sessions only.
        handler: List sessions handler (injected).

    Returns:
        SessionListResponse with list of sessions.
        JSONResponse with error on failure (401).
    """
    # Extract user_id and session_id from token
    user_id = await _extract_user_id_from_token(authorization, cache, db_session)
    current_session_id = await _extract_session_id_from_token(
        authorization, cache, db_session
    )

    if user_id is None:
        return _unauthorized_response(request)

    # Create query
    query = ListUserSessions(
        user_id=user_id,
        active_only=active_only,
        current_session_id=current_session_id,
    )

    # Execute handler
    result = await handler.handle(query)

    # Handle result (queries always succeed)
    match result:
        case Success(value=list_result):
            return SessionListResponse(
                sessions=[
                    SessionResponse(
                        id=s.id,
                        device_info=s.device_info,
                        ip_address=s.ip_address,
                        location=s.location,
                        created_at=s.created_at,
                        last_activity_at=s.last_activity_at,
                        expires_at=s.expires_at,
                        is_current=s.is_current,
                        is_revoked=s.is_revoked,
                    )
                    for s in list_result.sessions
                ],
                total_count=list_result.total_count,
                active_count=list_result.active_count,
            )
        case _:
            return _unauthorized_response(request)

get_session async

get_session(
    request: Request,
    session_id: UUID = Path(description="Session ID"),
    authorization: Annotated[str | None, Header()] = None,
    handler: GetSessionHandler = Depends(
        handler_factory(GetSessionHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionResponse | JSONResponse

Get details of a specific session.

GET /api/v1/sessions/{id} → 200 OK

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
session_id	UUID	Session ID from URL path.	Path(description='Session ID')
authorization	Annotated[str | None, Header()]	JWT access token from Authorization header.	None
handler	GetSessionHandler	Get session handler (injected).	Depends(handler_factory(GetSessionHandler))

Returns:

Type	Description
SessionResponse | JSONResponse	SessionResponse with session details.
SessionResponse | JSONResponse	JSONResponse with error on failure (401/404).

Source code in src/presentation/routers/api/v1/sessions.py

async def get_session(
    request: Request,
    session_id: UUID = Path(description="Session ID"),
    authorization: Annotated[str | None, Header()] = None,
    handler: GetSessionHandler = Depends(handler_factory(GetSessionHandler)),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionResponse | JSONResponse:
    """Get details of a specific session.

    GET /api/v1/sessions/{id} → 200 OK

    Args:
        request: FastAPI request object.
        session_id: Session ID from URL path.
        authorization: JWT access token from Authorization header.
        handler: Get session handler (injected).

    Returns:
        SessionResponse with session details.
        JSONResponse with error on failure (401/404).
    """
    # Extract user_id and current session_id from token
    user_id = await _extract_user_id_from_token(authorization, cache, db_session)
    current_session_id = await _extract_session_id_from_token(
        authorization, cache, db_session
    )

    if user_id is None:
        return _unauthorized_response(request)

    # Create query
    query = GetSession(
        session_id=session_id,
        user_id=user_id,
    )

    # Execute handler
    result = await handler.handle(query)

    # Handle result
    match result:
        case Success(value=session_result):
            return SessionResponse(
                id=session_result.id,
                device_info=session_result.device_info,
                ip_address=session_result.ip_address,
                location=session_result.location,
                created_at=session_result.created_at,
                last_activity_at=session_result.last_activity_at,
                expires_at=session_result.expires_at,
                is_current=session_result.id == current_session_id,
                is_revoked=session_result.is_revoked,
            )
        case Failure(error=error):
            if error == "session_not_found" or error == "not_session_owner":
                return _not_found_response(request, "Session not found")
            return _unauthorized_response(request)

revoke_session async

revoke_session(
    request: Request,
    session_id: UUID = Path(
        description="Session ID to revoke"
    ),
    data: SessionRevokeRequest | None = None,
    authorization: Annotated[str | None, Header()] = None,
    handler: RevokeSessionHandler = Depends(
        handler_factory(RevokeSessionHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> Response | JSONResponse

Revoke a specific session.

DELETE /api/v1/sessions/{id} → 204 No Content

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
session_id	UUID	Session ID from URL path.	Path(description='Session ID to revoke')
data	SessionRevokeRequest | None	Optional request body with reason.	None
authorization	Annotated[str | None, Header()]	JWT access token from Authorization header.	None
handler	RevokeSessionHandler	Revoke session handler (injected).	Depends(handler_factory(RevokeSessionHandler))

Returns:

Type	Description
Response | JSONResponse	204 No Content on success.
Response | JSONResponse	JSONResponse with error on failure (401/404).

Source code in src/presentation/routers/api/v1/sessions.py

async def revoke_session(
    request: Request,
    session_id: UUID = Path(description="Session ID to revoke"),
    data: SessionRevokeRequest | None = None,
    authorization: Annotated[str | None, Header()] = None,
    handler: RevokeSessionHandler = Depends(handler_factory(RevokeSessionHandler)),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> Response | JSONResponse:
    """Revoke a specific session.

    DELETE /api/v1/sessions/{id} → 204 No Content

    Args:
        request: FastAPI request object.
        session_id: Session ID from URL path.
        data: Optional request body with reason.
        authorization: JWT access token from Authorization header.
        handler: Revoke session handler (injected).

    Returns:
        204 No Content on success.
        JSONResponse with error on failure (401/404).
    """
    # Extract user_id from token
    user_id = await _extract_user_id_from_token(authorization, cache, db_session)

    if user_id is None:
        return _unauthorized_response(request)

    # Create command
    reason = data.reason if data else "manual"
    command = RevokeSession(
        session_id=session_id,
        user_id=user_id,
        reason=reason,
    )

    # Execute handler
    result = await handler.handle(command)

    # Handle result
    match result:
        case Success(value=_):
            return Response(status_code=status.HTTP_204_NO_CONTENT)
        case Failure(error=error):
            if error == "session_not_found" or error == "not_session_owner":
                return _not_found_response(request, "Session not found")
            if error == "session_already_revoked":
                return Response(status_code=status.HTTP_204_NO_CONTENT)
            return _unauthorized_response(request)

revoke_all_sessions async

revoke_all_sessions(
    request: Request,
    data: SessionRevokeAllRequest | None = None,
    authorization: Annotated[str | None, Header()] = None,
    handler: RevokeAllSessionsHandler = Depends(
        handler_factory(RevokeAllSessionsHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionRevokeAllResponse | JSONResponse

Revoke all sessions except current.

DELETE /api/v1/sessions → 200 OK

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
data	SessionRevokeAllRequest | None	Optional request body with reason.	None
authorization	Annotated[str | None, Header()]	JWT access token from Authorization header.	None
handler	RevokeAllSessionsHandler	Revoke all sessions handler (injected).	Depends(handler_factory(RevokeAllSessionsHandler))

Returns:

Type	Description
SessionRevokeAllResponse | JSONResponse	SessionRevokeAllResponse with count of revoked sessions.
SessionRevokeAllResponse | JSONResponse	JSONResponse with error on failure (401).

Source code in src/presentation/routers/api/v1/sessions.py

async def revoke_all_sessions(
    request: Request,
    data: SessionRevokeAllRequest | None = None,
    authorization: Annotated[str | None, Header()] = None,
    handler: RevokeAllSessionsHandler = Depends(
        handler_factory(RevokeAllSessionsHandler)
    ),
    cache: CacheProtocol = Depends(get_cache),
    db_session: AsyncSession = Depends(get_db_session),
) -> SessionRevokeAllResponse | JSONResponse:
    """Revoke all sessions except current.

    DELETE /api/v1/sessions → 200 OK

    Args:
        request: FastAPI request object.
        data: Optional request body with reason.
        authorization: JWT access token from Authorization header.
        handler: Revoke all sessions handler (injected).

    Returns:
        SessionRevokeAllResponse with count of revoked sessions.
        JSONResponse with error on failure (401).
    """
    # Extract user_id and session_id from token
    user_id = await _extract_user_id_from_token(authorization, cache, db_session)
    current_session_id = await _extract_session_id_from_token(
        authorization, cache, db_session
    )

    if user_id is None:
        return _unauthorized_response(request)

    # Create command
    reason = data.reason if data else "logout_all"
    command = RevokeAllUserSessions(
        user_id=user_id,
        reason=reason,
        except_session_id=current_session_id,
    )

    # Execute handler
    result = await handler.handle(command)

    # Handle result
    match result:
        case Success(value=revoked_count):
            return SessionRevokeAllResponse(
                revoked_count=revoked_count,
                message=f"{revoked_count} session(s) revoked successfully",
            )
        case _:
            return _unauthorized_response(request)