refactoring

Author: rafsaf

fastapi_plan/template/{{cookiecutter.project_name}}/app/api/routers/login.py

@@ -1,6 +1,4 @@
-from app.schemas.user import UserPydantic
 from datetime import timedelta
-from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.security import OAuth2PasswordRequestForm
@@ -15,7 +13,7 @@
 
 
 @router.post("/access-token", response_model=schemas.Token)
-async def login_access_token(form_data: OAuth2PasswordRequestForm = Depends()) -> Any:
+async def login_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
     """
     OAuth2 compatible token login, get an access token for future requests
     """
@@ -37,8 +35,8 @@ async def login_access_token(form_data: OAuth2PasswordRequestForm = Depends()) -
 
 
 @router.post("/test-token", response_model=schemas.UserPydantic)
-async def test_token(current_user: models.User = Depends(deps.get_current_user)) -> Any:
+async def test_token(current_user: models.User = Depends(deps.get_current_user)):
     """
     Test access token
     """
-    return UserPydantic.from_orm(current_user)
+    return await schemas.UserPydantic.from_tortoise_orm(current_user)

fastapi_plan/template/{{cookiecutter.project_name}}/app/api/routers/users.py

@@ -1,113 +1,115 @@
-from app.schemas.user import UserPydanticList
-from typing import Any, List
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, status
 from app import crud, models, schemas
-from app.schemas import UserPydantic
 from app.api import deps
 
 router = APIRouter()
 
 
-@router.get("/", response_model=UserPydanticList)
+@router.get("/", response_model=schemas.UserPydanticList)
 async def read_users(
     skip: int = 0,
     limit: int = 100,
     current_user: models.User = Depends(deps.get_current_active_superuser),
-) -> Any:
+):
     """
     Retrieve users.
     """
     users = await crud.user.get_multi(skip=skip, limit=limit)
-    return await UserPydanticList.from_queryset(users)
+    return await schemas.UserPydanticList.from_queryset(users)
 
 
-@router.post("/", response_model=UserPydantic)
+@router.post("/", response_model=schemas.UserPydantic)
 async def create_user(
     user_in: schemas.UserCreateBySuperuser,
     current_user: models.User = Depends(deps.get_current_active_superuser),
-) -> Any:
+):
     """
     Create new user.
     """
     user = await crud.user.get_by_email(email=user_in.email)
     if user:
         raise HTTPException(
-            status_code=400,
+            status_code=status.HTTP_400_BAD_REQUEST,
             detail="The user with this username already exists in the system.",
         )
     user = await crud.user.create_by_superuser(obj_in=user_in)
-    return UserPydantic.from_orm(user)
+    return await schemas.UserPydantic.from_tortoise_orm(user)
 
 
-@router.put("/me", response_model=UserPydantic)
+@router.put("/me", response_model=schemas.UserPydantic)
 async def update_user_me(
     user_in: schemas.UserUpdateMe,
     current_user: models.User = Depends(deps.get_current_active_user),
-) -> Any:
+):
     """
     Update own user.
     """
     user = await crud.user.update_me(db_obj=current_user, obj_in=user_in)
-    return schemas.UserPydantic.from_orm(user)
+    return await schemas.UserPydantic.from_tortoise_orm(user)
 
 
-@router.get("/me", response_model=UserPydantic)
+@router.get("/me", response_model=schemas.UserPydantic)
 async def read_user_me(
     current_user: models.User = Depends(deps.get_current_active_user),
-) -> Any:
+):
     """
     Get current user.
     """
-    return UserPydantic.from_orm(current_user)
+    return await schemas.UserPydantic.from_tortoise_orm(current_user)
 
 
-@router.post("/open", response_model=UserPydantic)
+@router.post("/open", response_model=schemas.UserPydantic)
 async def create_user_open(user_in: schemas.UserCreateMe):
     """
     Create new user without the need to be logged in.
     """
     user = await crud.user.get_by_email(email=user_in.email)
     if user:
         raise HTTPException(
-            status_code=400,
+            status_code=status.HTTP_400_BAD_REQUEST,
             detail="The user with this username already exists in the system",
         )
     user = await crud.user.create_me(obj_in=user_in)
-    return UserPydantic.from_orm(user)
+    return await schemas.UserPydantic.from_tortoise_orm(user)
 
 
-@router.get("/{user_id}", response_model=UserPydantic)
+@router.get("/{user_id}", response_model=schemas.UserPydantic)
 async def read_user_by_id(
     user_id: int,
     current_user: models.User = Depends(deps.get_current_active_user),
-) -> Any:
+):
     """
     Get a specific user by id.
     """
     user = await crud.user.get(id=user_id)
     if user == current_user:
         return user
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail="The user does not exist"
+        )
     if not crud.user.is_superuser(current_user):
         raise HTTPException(
-            status_code=400, detail="The user doesn't have enough privileges"
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="The user doesn't have enough privileges",
         )
-    return UserPydantic.from_orm(user)
+    return await schemas.UserPydantic.from_tortoise_orm(user)
 
 
-@router.put("/{user_id}", response_model=UserPydantic)
+@router.put("/{user_id}", response_model=schemas.UserPydantic)
 async def update_user(
     user_id: int,
     user_in: schemas.UserUpdateBySuperuser,
     current_user: models.User = Depends(deps.get_current_active_superuser),
-) -> Any:
+):
     """
     Update a user.
     """
     user = await crud.user.get(id=user_id)
     if not user:
         raise HTTPException(
-            status_code=404,
+            status_code=status.HTTP_404_NOT_FOUND,
             detail="The user with this username does not exist in the system",
         )
     user = await crud.user.update_by_superuser(db_obj=user, obj_in=user_in)
-    return UserPydantic.from_orm(user)
+    return await schemas.UserPydantic.from_tortoise_orm(user)

fastapi_plan/template/{{cookiecutter.project_name}}/app/crud/crud_user.py

@@ -1,40 +1,42 @@
-from app.schemas.user import UserCreateBySuperuser, UserUpdateBySuperuser
 from typing import Optional
 
 from app.core.security import get_password_hash, verify_password
 from app.crud.base import CRUDBase
-from app.models import User
-from app.schemas import UserCreateMe, UserUpdateMe
+from app import models, schemas
 from tortoise.exceptions import DoesNotExist
 
 
-class CRUDUser(CRUDBase[User, UserCreateMe, UserUpdateMe]):
-    async def get_by_email(self, email: str) -> Optional[User]:
+class CRUDUser(CRUDBase[models.User, schemas.UserCreateMe, schemas.UserUpdateMe]):
+    async def get_by_email(self, email: str) -> Optional[models.User]:
         try:
-            user = await User.get(email=email)
+            user = await models.User.get(email=email)
         except DoesNotExist:
             return None
         else:
             return user
 
-    async def create_me(self, obj_in: UserCreateMe) -> User:
-        db_obj = await User.create(
+    async def create_me(self, obj_in: schemas.UserCreateMe) -> models.User:
+        db_obj = await models.User.create(
             email=obj_in.email,
             password_hash=get_password_hash(obj_in.password),
         )
         return db_obj
 
-    async def create_by_superuser(self, obj_in: UserCreateBySuperuser) -> User:
+    async def create_by_superuser(
+        self, obj_in: schemas.UserCreateBySuperuser
+    ) -> models.User:
 
-        db_obj = await User.create(
+        db_obj = await models.User.create(
             email=obj_in.email,
             password_hash=get_password_hash(obj_in.password),
             is_superuser=obj_in.is_superuser,
             is_active=obj_in.is_active,
         )
         return db_obj
 
-    async def update_me(self, db_obj: User, obj_in: UserUpdateMe) -> User:
+    async def update_me(
+        self, db_obj: models.User, obj_in: schemas.UserUpdateMe
+    ) -> models.User:
         db_obj.name = obj_in.name
         db_obj.family_name = obj_in.family_name
         if obj_in.password:
@@ -45,8 +47,8 @@ async def update_me(self, db_obj: User, obj_in: UserUpdateMe) -> User:
         return db_obj
 
     async def update_by_superuser(
-        self, db_obj: User, obj_in: UserUpdateBySuperuser
-    ) -> User:
+        self, db_obj: models.User, obj_in: schemas.UserUpdateBySuperuser
+    ) -> models.User:
 
         db_obj.name = obj_in.name
         db_obj.family_name = obj_in.family_name
@@ -60,19 +62,19 @@ async def update_by_superuser(
         await db_obj.refresh_from_db()
         return db_obj
 
-    async def authenticate(self, email: str, password: str) -> Optional[User]:
+    async def authenticate(self, email: str, password: str) -> Optional[models.User]:
         user = await self.get_by_email(email=email)
         if not user:
             return None
         if not verify_password(password, user.password_hash):
             return None
         return user
 
-    def is_active(self, user: User) -> bool:
+    def is_active(self, user: models.User) -> bool:
         return user.is_active  # type: ignore
 
-    def is_superuser(self, user: User) -> bool:
+    def is_superuser(self, user: models.User) -> bool:
         return user.is_superuser  # type: ignore
 
 
-user = CRUDUser(User)
+user = CRUDUser(models.User)

fastapi_plan/template/{{cookiecutter.project_name}}/app/schemas/__init__.py

@@ -1,6 +1,7 @@
 from .token import Token, TokenPayload  # noqa
 from .user import (
     UserPydantic,
+    UserPydanticList,
     UserCreateMe,
     UserCreateBySuperuser,
     UserUpdateBySuperuser,