-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Showing
11 changed files
with
285 additions
and
112 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,90 +1,87 @@ | ||
| """Routes for users""" | ||
|
|
||
| from fastapi import APIRouter, Depends, HTTPException, Query | ||
| from fastapi import APIRouter, Depends, Query, HTTPException | ||
| from pydantic import EmailStr | ||
| from sqlmodel import Session | ||
| from sqlmodel.sql.expression import Select, SelectOfScalar | ||
| from starlette import status | ||
|
|
||
| from starlette.responses import JSONResponse | ||
|
|
||
| from genotype_api.database.crud.read import ( | ||
| get_user_by_id, | ||
| get_user_by_email, | ||
| get_users_with_skip_and_limit, | ||
| ) | ||
| from genotype_api.database.crud.update import update_user_email | ||
| from genotype_api.database.crud import delete, create | ||
| from genotype_api.database.models import User | ||
| from genotype_api.dto.dto import UserRead, UserCreate, UserReadWithPlates | ||
| from genotype_api.database.session_handler import get_session | ||
| from genotype_api.dto.user import UserRequest, UserResponse | ||
| from genotype_api.exceptions import UserNotFoundError, UserArchiveError, UserExistsError | ||
| from genotype_api.security import get_active_user | ||
| from genotype_api.services.user_service.user_service import UserService | ||
|
|
||
| SelectOfScalar.inherit_cache = True | ||
| Select.inherit_cache = True | ||
| router = APIRouter() | ||
|
|
||
|
|
||
| router = APIRouter() | ||
| def get_user_service(session: Session = Depends(get_session)) -> UserService: | ||
| return UserService(session) | ||
|
|
||
|
|
||
| @router.get("/{user_id}", response_model=UserReadWithPlates) | ||
| @router.get("/{user_id}", response_model=UserResponse) | ||
| def read_user( | ||
| user_id: int, | ||
| session: Session = Depends(get_session), | ||
| user_service: UserService = Depends(get_user_service), | ||
| current_user: User = Depends(get_active_user), | ||
| ) -> User: | ||
| return get_user_by_id(session=session, user_id=user_id) | ||
| ) -> UserResponse: | ||
| try: | ||
| return user_service.get_user(user_id) | ||
| except UserNotFoundError: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | ||
|
|
||
|
|
||
| @router.delete("/{user_id}") | ||
| def delete_user( | ||
| user_id: int, | ||
| session: Session = Depends(get_session), | ||
| user_service: UserService = Depends(get_user_service), | ||
| current_user: User = Depends(get_active_user), | ||
| ) -> JSONResponse: | ||
| user: User = get_user_by_id(session=session, user_id=user_id) | ||
| if not user: | ||
| try: | ||
| user_service.delete_user(user_id) | ||
| except UserNotFoundError: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | ||
| if user.plates: | ||
| raise HTTPException( | ||
| except UserArchiveError: | ||
| HTTPException( | ||
| status_code=status.HTTP_406_NOT_ACCEPTABLE, | ||
| detail="User previously signed plates, please archive instead", | ||
| ) | ||
| delete.delete_user(session=session, user=user) | ||
| return JSONResponse(content="User deleted successfully", status_code=status.HTTP_200_OK) | ||
| return JSONResponse(content=f"Deleted user with id: {user_id}.", status_code=status.HTTP_200_OK) | ||
|
|
||
|
|
||
| @router.put("/{user_id}/email", response_model=User) | ||
| @router.put("/{user_id}/email", response_model=UserResponse, response_model_exclude={"plates"}) | ||
| def change_user_email( | ||
| user_id: int, | ||
| email: EmailStr, | ||
| session: Session = Depends(get_session), | ||
| user_service: UserService = Depends(get_user_service), | ||
| current_user: User = Depends(get_active_user), | ||
| ) -> User: | ||
| user: User = get_user_by_id(session=session, user_id=user_id) | ||
| if not user: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | ||
| user: User = update_user_email(session=session, user=user, email=email) | ||
| return user | ||
| ) -> UserResponse: | ||
| try: | ||
| return user_service.update_user_email(user_id=user_id, email=email) | ||
| except UserNotFoundError: | ||
| HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") | ||
|
|
||
|
|
||
| @router.get("/", response_model=list[UserRead]) | ||
| @router.get("/", response_model=list[UserResponse], response_model_exclude={"plates"}) | ||
| def read_users( | ||
| skip: int = 0, | ||
| limit: int = Query(default=100, lte=100), | ||
| session: Session = Depends(get_session), | ||
| user_service: UserService = Depends(get_user_service), | ||
| current_user: User = Depends(get_active_user), | ||
| ) -> list[User]: | ||
| return get_users_with_skip_and_limit(session=session, skip=skip, limit=limit) | ||
| ) -> list[UserResponse]: | ||
|
|
||
| return user_service.get_users(skip=skip, limit=limit) | ||
|
|
||
|
|
||
| @router.post("/", response_model=User) | ||
| @router.post("/", response_model=UserResponse, response_model_exclude={"plates"}) | ||
| def create_user( | ||
| user: UserCreate, | ||
| session: Session = Depends(get_session), | ||
| user: UserRequest, | ||
| user_service: UserService = Depends(get_user_service), | ||
| current_user: User = Depends(get_active_user), | ||
| ): | ||
| existing_user: User = get_user_by_email(session=session, email=user.email) | ||
| if existing_user: | ||
| raise HTTPException(status_code=409, detail="Email already registered") | ||
| new_user: User = create.create_user(session=session, user=user) | ||
| return new_user | ||
| try: | ||
| return user_service.create_user(user) | ||
| except UserExistsError: | ||
| HTTPException(status_code=409, detail="Email already registered.") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,25 @@ | ||
| """Module for the plate DTOs.""" | ||
|
|
||
| from datetime import datetime | ||
|
|
||
| from pydantic import BaseModel, EmailStr | ||
|
|
||
|
|
||
| class PlateOnUser(BaseModel): | ||
| created_at: datetime | None = None | ||
| plate_id: str | None = None | ||
| signed_by: int | None = None | ||
| signed_at: datetime | None = None | ||
| id: str | None = None | ||
|
|
||
|
|
||
| class UserResponse(BaseModel): | ||
| email: EmailStr | None = None | ||
| name: str | None = None | ||
| id: int | None = None | ||
| plates: list[PlateOnUser] | None = None | ||
|
|
||
|
|
||
| class UserRequest(BaseModel): | ||
| email: EmailStr | ||
| name: str |
Oops, something went wrong.