39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
from typing import Annotated
|
|
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
|
|
from ..model.token import RefreshTokenPayload, Token
|
|
from ..context import ContextDependency, Context
|
|
from ..utils.jwt import AccessToken, RefreshToken
|
|
from ..domain import User
|
|
|
|
router = APIRouter(tags=["token"])
|
|
|
|
|
|
@router.post("", response_model=Token)
|
|
async def create_token(
|
|
data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
|
ctx: Context = Depends(ContextDependency()),
|
|
):
|
|
user = await ctx.repo.user.get(User.username == data.username)
|
|
|
|
if user is None or not user.verify_password(data.password):
|
|
raise HTTPException(status_code=401, detail="Unathorized")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
access_token = AccessToken(sub=user.id, exp=AccessToken.calculate_exp(now))
|
|
refresh_token = RefreshToken(sub=user.id, exp=RefreshToken.calculate_exp(now))
|
|
return Token(
|
|
access_token=access_token.encode(),
|
|
refresh_token=refresh_token.encode(),
|
|
)
|
|
|
|
|
|
@router.post("/refresh", response_model=Token)
|
|
async def refresh_token(
|
|
data: RefreshTokenPayload,
|
|
ctx: Context = Depends(ContextDependency()),
|
|
):
|
|
pass
|