42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from datetime import datetime
|
|
|
|
from fooder.model.token import TokenResponse
|
|
from fooder.context import ContextDependency, Context
|
|
from fooder.controller.user import UserController
|
|
from fooder.utils.jwt import RefreshToken, generate_token_pair
|
|
|
|
router = APIRouter(tags=["token"])
|
|
|
|
_ctx = ContextDependency()
|
|
|
|
|
|
def gen_token_response(entity_id: int, now: datetime) -> TokenResponse:
|
|
access_token, refresh_token = generate_token_pair(entity_id, now)
|
|
return TokenResponse(
|
|
access_token=access_token.encode(),
|
|
refresh_token=refresh_token.encode(),
|
|
)
|
|
|
|
|
|
@router.post("", response_model=TokenResponse)
|
|
async def token_create(
|
|
data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
|
ctx: Context = Depends(_ctx),
|
|
) -> TokenResponse:
|
|
now = ctx.clock()
|
|
user_ctrl = await UserController.session_start(ctx, data.username, data.password)
|
|
return gen_token_response(user_ctrl.obj.id, now)
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
async def token_refresh(
|
|
refresh_token: str,
|
|
ctx: Context = Depends(_ctx),
|
|
) -> TokenResponse:
|
|
now = ctx.clock()
|
|
token = RefreshToken.decode(refresh_token)
|
|
return gen_token_response(token.sub, now)
|