from typing import Any from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app import crud, models, schemas from app.api import deps router = APIRouter() @router.post("/", response_model=schemas.User, status_code=status.HTTP_201_CREATED) def create_user( *, db: Session = Depends(deps.get_db), user_in: schemas.UserCreate, ) -> Any: """ Create new user. """ user = crud.user.get_by_email(db, email=user_in.email) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="The user with this email already exists in the system.", ) user = crud.user.get_by_username(db, username=user_in.username) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="The user with this username already exists in the system.", ) user = crud.user.create(db, obj_in=user_in) return user @router.get("/me", response_model=schemas.User) def read_user_me( current_user: models.User = Depends(deps.get_current_user), ) -> Any: """ Get current user. """ return current_user @router.put("/me", response_model=schemas.User) def update_user_me( *, db: Session = Depends(deps.get_db), user_in: schemas.UserUpdate, current_user: models.User = Depends(deps.get_current_user), ) -> Any: """ Update own user. """ user = crud.user.update(db, db_obj=current_user, obj_in=user_in) return user