Skip to content

presentation.routers.api.v1.tokens

src.presentation.routers.api.v1.tokens

Tokens resource handlers.

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

Handlers

create_tokens - Create new tokens (refresh)

Classes

Functions

create_tokens async 

create_tokens(
    request: Request,
    data: TokenCreateRequest,
    handler: RefreshAccessTokenHandler = Depends(
        handler_factory(RefreshAccessTokenHandler)
    ),
) -> TokenCreateResponse | JSONResponse

Create new tokens (refresh).

POST /api/v1/tokens → 201 Created

Exchanges a valid refresh token for new access and refresh tokens. Implements token rotation: old refresh token is invalidated.

Parameters:

Name Type Description Default
request Request

FastAPI request object.

 required
data TokenCreateRequest

Token creation request (refresh_token).

 required
handler RefreshAccessTokenHandler

Refresh token handler (injected).

 Depends(handler_factory(RefreshAccessTokenHandler))

Returns:

Type Description
TokenCreateResponse | JSONResponse

TokenCreateResponse on success (201 Created).
TokenCreateResponse | JSONResponse

JSONResponse with error on failure (400/401).
Source code in src/presentation/routers/api/v1/tokens.py 
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
async def create_tokens(
    request: Request,
    data: TokenCreateRequest,
    handler: RefreshAccessTokenHandler = Depends(
        handler_factory(RefreshAccessTokenHandler)
    ),
) -> TokenCreateResponse | JSONResponse:
    """Create new tokens (refresh).

    POST /api/v1/tokens → 201 Created

    Exchanges a valid refresh token for new access and refresh tokens.
    Implements token rotation: old refresh token is invalidated.

    Args:
        request: FastAPI request object.
        data: Token creation request (refresh_token).
        handler: Refresh token handler (injected).

    Returns:
        TokenCreateResponse on success (201 Created).
        JSONResponse with error on failure (400/401).
    """
    # Create command from request data
    command = RefreshAccessToken(
        refresh_token=data.refresh_token,
    )

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

    # Handle result
    match result:
        case Success(value=refresh_response):
            return TokenCreateResponse(
                access_token=refresh_response.access_token,
                refresh_token=refresh_response.refresh_token,
                token_type=refresh_response.token_type,
                expires_in=refresh_response.expires_in,
            )
        case Failure(error=error):
            # Map error to ApplicationErrorCode
            error_mapping = {
                "token_invalid": ApplicationErrorCode.COMMAND_VALIDATION_FAILED,
                "token_expired": ApplicationErrorCode.UNAUTHORIZED,
                "token_revoked": ApplicationErrorCode.UNAUTHORIZED,
                "user_not_found": ApplicationErrorCode.UNAUTHORIZED,
                "user_inactive": ApplicationErrorCode.UNAUTHORIZED,
            }
            error_code = error_mapping.get(
                error, ApplicationErrorCode.COMMAND_VALIDATION_FAILED
            )

            app_error = ApplicationError(
                code=error_code,
                message=_get_user_friendly_error(error),
            )
            return ErrorResponseBuilder.from_application_error(
                error=app_error,
                request=request,
                trace_id=get_trace_id() or "",
            )