107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from typing import Annotated, Union
|
|
from sqlalchemy.orm import Session
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Response
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from io import BytesIO
|
|
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from ...config import jwt_config
|
|
from ...dependencies import get_db
|
|
from . import schemas
|
|
from . import services
|
|
|
|
|
|
router = APIRouter(
|
|
prefix="/auth",
|
|
tags=["auth"],
|
|
dependencies=[Depends(get_db)],
|
|
responses={404: {"description": "Not found"}},
|
|
)
|
|
|
|
|
|
@router.post("/token")
|
|
async def login_for_access_token(
|
|
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> schemas.Token:
|
|
user = services.authenticate_user(db, form_data.username, form_data.password)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
access_token_expires = timedelta(weeks=jwt_config.ACCESS_TOKEN_EXPIRE_WEEKS)
|
|
access_token = services.create_access_token(
|
|
data={"sub": user.username}, expires_delta=access_token_expires
|
|
)
|
|
return schemas.Token(access_token=access_token, token_type="bearer")
|
|
|
|
|
|
@router.post("/register")
|
|
async def register(
|
|
user_data: schemas.UserRegister,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> schemas.User:
|
|
if services.check_captcha(
|
|
id=user_data.captcha.id, prompt=user_data.captcha.prompt, db=db
|
|
):
|
|
user = services.get_user_by_username(db, user_data.username)
|
|
if user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="User with this username already exists",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
if user_data.password != user_data.password2:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Passwords do not match",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
user = services.create_user(db=db, user_data=user_data)
|
|
return user
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid captcha"
|
|
)
|
|
|
|
|
|
@router.get("/me")
|
|
async def read_users_me(
|
|
current_user: Annotated[
|
|
schemas.UserInDB, Depends(services.get_current_active_user)
|
|
],
|
|
) -> schemas.UserInDB:
|
|
return current_user
|
|
|
|
|
|
@router.get("/anon")
|
|
async def get_anon_user(
|
|
anon_user: Annotated[schemas.AnonUser, Depends(services.get_anon_user)]
|
|
) -> schemas.AnonUser:
|
|
return anon_user
|
|
|
|
|
|
@router.patch("/anon")
|
|
async def get_anon_user(
|
|
anon_user: Annotated[schemas.AnonUser, Depends(services.patch_anon_name)]
|
|
) -> schemas.AnonUser:
|
|
return anon_user
|
|
|
|
|
|
@router.get(
|
|
"/captcha/{captcha_id}",
|
|
responses={200: {"content": {"image/png": {}}}},
|
|
response_class=Response,
|
|
)
|
|
async def generate_captcha(
|
|
captcha: Annotated[BytesIO, Depends(services.get_captcha)]
|
|
) -> Response:
|
|
captcha.seek(0)
|
|
captcha_bytes = captcha.read()
|
|
return Response(content=captcha_bytes, media_type="image/png")
|