Done setup template
This commit is contained in:
9
backend/routes/__init__.py
Normal file
9
backend/routes/__init__.py
Normal file
@ -0,0 +1,9 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
|
||||
from . import (auth, user)
|
||||
|
||||
router = APIRouter(prefix='/api')
|
||||
|
||||
router.include_router(auth.router)
|
||||
router.include_router(user.router)
|
9
backend/routes/_base/routers.py
Normal file
9
backend/routes/_base/routers.py
Normal file
@ -0,0 +1,9 @@
|
||||
from enum import Enum
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from backend.core.dependencies import get_auth_user
|
||||
|
||||
|
||||
class PrivateAPIRouter(APIRouter):
|
||||
def __init__(self, tags: list[str | Enum] | None = None, prefix: str = "", **kwargs):
|
||||
super().__init__(tags=tags, prefix=prefix, dependencies=[Depends(get_auth_user)], **kwargs)
|
7
backend/routes/auth/__init__.py
Normal file
7
backend/routes/auth/__init__.py
Normal file
@ -0,0 +1,7 @@
|
||||
|
||||
from fastapi import APIRouter
|
||||
from . import auth
|
||||
|
||||
router = APIRouter(prefix='/auth')
|
||||
|
||||
router.include_router(auth.auth_router)
|
70
backend/routes/auth/auth.py
Normal file
70
backend/routes/auth/auth.py
Normal file
@ -0,0 +1,70 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Annotated, Any
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response, status
|
||||
|
||||
# from fastapi.encoders import jsonable_encoder
|
||||
from fastapi.security import OAuth2PasswordRequestForm
|
||||
from sqlalchemy.orm import Session
|
||||
from backend.core.config import get_app_settings
|
||||
from backend.core.dependencies.dependencies import get_current_user
|
||||
from backend.core import MessageCode
|
||||
from backend.db.db_setup import generate_session
|
||||
from backend.schemas import ReturnValue, UserRequest, LoginResponse, UserCreate, PrivateUser
|
||||
from backend.services.user import UserService
|
||||
|
||||
|
||||
auth_router = APIRouter(tags=["Users: Authentication"])
|
||||
user_service = UserService()
|
||||
settings = get_app_settings()
|
||||
|
||||
db_dependency = Annotated[Session, Depends(generate_session)]
|
||||
current_user_token = Annotated[PrivateUser, Depends(get_current_user)]
|
||||
|
||||
@auth_router.post('/token')
|
||||
async def get_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], db: db_dependency):
|
||||
user = user_service.check_exist(user=UserRequest(username=form_data.username, password=form_data.password))
|
||||
if not user:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=MessageCode.WRONG_INPUT)
|
||||
token = user_service.get_access_token(user_id=user.id)
|
||||
return {'access_token': token, 'token_type': 'bearer'}
|
||||
|
||||
|
||||
@auth_router.put('/register')
|
||||
def register_user(user: UserCreate, db: db_dependency) -> ReturnValue[Any]:
|
||||
db_user = user_service.get_by_username(username=user.username)
|
||||
if db_user:
|
||||
raise HTTPException(status_code=400, detail=MessageCode.CREATED_USER)
|
||||
user_service.create(db=db, user=user)
|
||||
return ReturnValue(status=200, data="created")
|
||||
|
||||
@auth_router.post('/login', response_model=ReturnValue[LoginResponse])
|
||||
def user_login(user: UserRequest, response: Response) -> ReturnValue[Any]:
|
||||
db_user = user_service.check_exist(user=user)
|
||||
if not db_user:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=MessageCode.WRONG_INPUT)
|
||||
if db_user.is_lock is True:
|
||||
raise HTTPException(status_code=status.HTTP_423_LOCKED, detail=MessageCode.ACCOUNT_LOCK)
|
||||
access_token, refresh_token = user_service.generate_token(user_id=db_user.id)
|
||||
duration_access = datetime.now(timezone.utc) + timedelta(minutes=settings.EXP_TOKEN)
|
||||
duration_refresh = int(timedelta(days=settings.EXP_REFRESH).total_seconds())
|
||||
response.set_cookie(
|
||||
key=settings.COOKIE_KEY,
|
||||
value=refresh_token,
|
||||
max_age=duration_refresh,
|
||||
expires=duration_refresh,
|
||||
httponly=True,
|
||||
samesite="strict",
|
||||
)
|
||||
return ReturnValue(status=200, data=dict(access_token=access_token, exp=int(duration_access.timestamp()), name=db_user.name))
|
||||
|
||||
@auth_router.get('/refresh')
|
||||
def user_check(current_user: current_user_token) -> ReturnValue[Any]:
|
||||
access_token = user_service.get_access_token(user_id=current_user.id)
|
||||
duration_access = datetime.now(timezone.utc) + timedelta(minutes=settings.EXP_TOKEN)
|
||||
return ReturnValue(status=200, data=dict(accessToken=access_token, exp=int(duration_access.timestamp())))
|
||||
|
||||
@auth_router.get('/logout')
|
||||
def user_logout(response: Response, current_user: current_user_token) -> ReturnValue[Any]:
|
||||
if current_user:
|
||||
response.delete_cookie(key=settings.COOKIE_KEY)
|
||||
return ReturnValue(status=200, data='Logged out')
|
7
backend/routes/user/__init__.py
Normal file
7
backend/routes/user/__init__.py
Normal file
@ -0,0 +1,7 @@
|
||||
|
||||
from fastapi import APIRouter
|
||||
from . import user
|
||||
|
||||
router = APIRouter(prefix='/user')
|
||||
|
||||
router.include_router(user.public_router)
|
21
backend/routes/user/user.py
Normal file
21
backend/routes/user/user.py
Normal file
@ -0,0 +1,21 @@
|
||||
from typing import Annotated, Any
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy.orm import Session
|
||||
from backend.core.config import get_app_settings
|
||||
from backend.core.dependencies import is_logged_in
|
||||
from backend.db.db_setup import generate_session
|
||||
from backend.schemas.common import ReturnValue
|
||||
from backend.schemas.user import ProfileResponse
|
||||
from backend.services.user import UserService
|
||||
|
||||
|
||||
public_router = APIRouter(tags=["Users: Info"])
|
||||
user_service = UserService()
|
||||
settings = get_app_settings()
|
||||
|
||||
db_dependency = Annotated[Session, Depends(generate_session)]
|
||||
current_user_token = Annotated[ProfileResponse, Depends(is_logged_in)]
|
||||
|
||||
@public_router.get("/me", response_model=ReturnValue[ProfileResponse])
|
||||
def get_user(current_user: current_user_token) -> ReturnValue[Any]:
|
||||
return ReturnValue(status=200, data=current_user)
|
Reference in New Issue
Block a user