Refactor backend, settings, DB sessions, types, configs, plugins

{{cookiecutter.project_slug}}/backend/app/.flake8

@@ -0,0 +1,3 @@
+[flake8]
+max-line-length = 88
+exclude = .git,__pycache__,__init__.py,.mypy_cache,.pytest_cache

{{cookiecutter.project_slug}}/backend/app/.gitignore

@@ -0,0 +1,3 @@
+.mypy_cache
+.coverage
+htmlcov

{{cookiecutter.project_slug}}/backend/app/app/api/api_v1/endpoints/items.py

@@ -1,104 +1,99 @@
-from typing import List
+from typing import Any, List
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 
-from app import crud
-from app.api.utils.db import get_db
-from app.api.utils.security import get_current_active_user
-from app.models.user import User as DBUser
-from app.schemas.item import Item, ItemCreate, ItemUpdate
+from app import crud, models, schemas
+from app.api import deps
 
 router = APIRouter()
 
 
-@router.get("/", response_model=List[Item])
+@router.get("/", response_model=List[schemas.Item])
 def read_items(
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     skip: int = 0,
     limit: int = 100,
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Retrieve items.
     """
     if crud.user.is_superuser(current_user):
         items = crud.item.get_multi(db, skip=skip, limit=limit)
     else:
         items = crud.item.get_multi_by_owner(
-            db_session=db, owner_id=current_user.id, skip=skip, limit=limit
+            db=db, owner_id=current_user.id, skip=skip, limit=limit
         )
     return items
 
 
-@router.post("/", response_model=Item)
+@router.post("/", response_model=schemas.Item)
 def create_item(
     *,
-    db: Session = Depends(get_db),
-    item_in: ItemCreate,
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    db: Session = Depends(deps.get_db),
+    item_in: schemas.ItemCreate,
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Create new item.
     """
-    item = crud.item.create_with_owner(
-        db_session=db, obj_in=item_in, owner_id=current_user.id
-    )
+    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=current_user.id)
     return item
 
 
-@router.put("/{id}", response_model=Item)
+@router.put("/{id}", response_model=schemas.Item)
 def update_item(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     id: int,
-    item_in: ItemUpdate,
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    item_in: schemas.ItemUpdate,
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Update an item.
     """
-    item = crud.item.get(db_session=db, id=id)
+    item = crud.item.get(db=db, id=id)
     if not item:
         raise HTTPException(status_code=404, detail="Item not found")
     if not crud.user.is_superuser(current_user) and (item.owner_id != current_user.id):
         raise HTTPException(status_code=400, detail="Not enough permissions")
-    item = crud.item.update(db_session=db, db_obj=item, obj_in=item_in)
+    item = crud.item.update(db=db, db_obj=item, obj_in=item_in)
     return item
 
 
-@router.get("/{id}", response_model=Item)
+@router.get("/{id}", response_model=schemas.Item)
 def read_item(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     id: int,
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Get item by ID.
     """
-    item = crud.item.get(db_session=db, id=id)
+    item = crud.item.get(db=db, id=id)
     if not item:
         raise HTTPException(status_code=404, detail="Item not found")
     if not crud.user.is_superuser(current_user) and (item.owner_id != current_user.id):
         raise HTTPException(status_code=400, detail="Not enough permissions")
     return item
 
 
-@router.delete("/{id}", response_model=Item)
+@router.delete("/{id}", response_model=schemas.Item)
 def delete_item(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     id: int,
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Delete an item.
     """
-    item = crud.item.get(db_session=db, id=id)
+    item = crud.item.get(db=db, id=id)
     if not item:
         raise HTTPException(status_code=404, detail="Item not found")
     if not crud.user.is_superuser(current_user) and (item.owner_id != current_user.id):
         raise HTTPException(status_code=400, detail="Not enough permissions")
-    item = crud.item.remove(db_session=db, id=id)
+    item = crud.item.remove(db=db, id=id)
     return item

{{cookiecutter.project_slug}}/backend/app/app/api/api_v1/endpoints/#.py

@@ -1,19 +1,15 @@
 from datetime import timedelta
+from typing import Any
 
 from fastapi import APIRouter, Body, Depends, HTTPException
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.orm import Session
 
-from app import crud
-from app.api.utils.db import get_db
-from app.api.utils.security import get_current_user
+from app import crud, models, schemas
+from app.api import deps
+from app.core import security
 from app.core.config import settings
-from app.core.jwt import create_access_token
 from app.core.security import get_password_hash
-from app.models.user import User as DBUser
-from app.schemas.msg import Msg
-from app.schemas.token import Token
-from app.schemas.user import User
 from app.utils import (
     generate_password_reset_token,
     send_reset_password_email,
@@ -23,10 +19,10 @@
 router = APIRouter()
 
 
-@router.post("/#/access-token", response_model=Token)
+@router.post("/#/access-token", response_model=schemas.Token)
 def login_access_token(
-    db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
-):
+    db: Session = Depends(deps.get_db), form_data: OAuth2PasswordRequestForm = Depends()
+) -> Any:
     """
     OAuth2 compatible token login, get an access token for future requests
     """
@@ -39,23 +35,23 @@ def login_access_token(
         raise HTTPException(status_code=400, detail="Inactive user")
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     return {
-        "access_token": create_access_token(
-            data={"user_id": user.id}, expires_delta=access_token_expires
+        "access_token": security.create_access_token(
+            user.id, expires_delta=access_token_expires
         ),
         "token_type": "bearer",
     }
 
 
-@router.post("/#/test-token", response_model=User)
-def test_token(current_user: DBUser = Depends(get_current_user)):
+@router.post("/#/test-token", response_model=schemas.User)
+def test_token(current_user: models.User = Depends(deps.get_current_user)) -> Any:
     """
     Test access token
     """
     return current_user
 
 
-@router.post("/password-recovery/{email}", response_model=Msg)
-def recover_password(email: str, db: Session = Depends(get_db)):
+@router.post("/password-recovery/{email}", response_model=schemas.Msg)
+def recover_password(email: str, db: Session = Depends(deps.get_db)) -> Any:
     """
     Password Recovery
     """
@@ -73,10 +69,12 @@ def recover_password(email: str, db: Session = Depends(get_db)):
     return {"msg": "Password recovery email sent"}
 
 
-@router.post("/reset-password/", response_model=Msg)
+@router.post("/reset-password/", response_model=schemas.Msg)
 def reset_password(
-    token: str = Body(...), new_password: str = Body(...), db: Session = Depends(get_db)
-):
+    token: str = Body(...),
+    new_password: str = Body(...),
+    db: Session = Depends(deps.get_db),
+) -> Any:
     """
     Reset password
     """

{{cookiecutter.project_slug}}/backend/app/app/api/api_v1/endpoints/users.py

@@ -1,42 +1,39 @@
-from typing import List
+from typing import Any, List
 
 from fastapi import APIRouter, Body, Depends, HTTPException
 from fastapi.encoders import jsonable_encoder
 from pydantic.networks import EmailStr
 from sqlalchemy.orm import Session
 
-from app import crud
-from app.api.utils.db import get_db
-from app.api.utils.security import get_current_active_superuser, get_current_active_user
+from app import crud, models, schemas
+from app.api import deps
 from app.core.config import settings
-from app.models.user import User as DBUser
-from app.schemas.user import User, UserCreate, UserUpdate
 from app.utils import send_new_account_email
 
 router = APIRouter()
 
 
-@router.get("/", response_model=List[User])
+@router.get("/", response_model=List[schemas.User])
 def read_users(
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     skip: int = 0,
     limit: int = 100,
-    current_user: DBUser = Depends(get_current_active_superuser),
-):
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
     """
     Retrieve users.
     """
     users = crud.user.get_multi(db, skip=skip, limit=limit)
     return users
 
 
-@router.post("/", response_model=User)
+@router.post("/", response_model=schemas.User)
 def create_user(
     *,
-    db: Session = Depends(get_db),
-    user_in: UserCreate,
-    current_user: DBUser = Depends(get_current_active_superuser),
-):
+    db: Session = Depends(deps.get_db),
+    user_in: schemas.UserCreate,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
     """
     Create new user.
     """
@@ -54,20 +51,20 @@ def create_user(
     return user
 
 
-@router.put("/me", response_model=User)
+@router.put("/me", response_model=schemas.User)
 def update_user_me(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     password: str = Body(None),
     full_name: str = Body(None),
     email: EmailStr = Body(None),
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Update own user.
     """
     current_user_data = jsonable_encoder(current_user)
-    user_in = UserUpdate(**current_user_data)
+    user_in = schemas.UserUpdate(**current_user_data)
     if password is not None:
         user_in.password = password
     if full_name is not None:
@@ -78,25 +75,25 @@ def update_user_me(
     return user
 
 
-@router.get("/me", response_model=User)
+@router.get("/me", response_model=schemas.User)
 def read_user_me(
-    db: Session = Depends(get_db),
-    current_user: DBUser = Depends(get_current_active_user),
-):
+    db: Session = Depends(deps.get_db),
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
     """
     Get current user.
     """
     return current_user
 
 
-@router.post("/open", response_model=User)
+@router.post("/open", response_model=schemas.User)
 def create_user_open(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     password: str = Body(...),
     email: EmailStr = Body(...),
     full_name: str = Body(None),
-):
+) -> Any:
     """
     Create new user without the need to be logged in.
     """
@@ -111,17 +108,17 @@ def create_user_open(
             status_code=400,
             detail="The user with this username already exists in the system",
         )
-    user_in = UserCreate(password=password, email=email, full_name=full_name)
+    user_in = schemas.UserCreate(password=password, email=email, full_name=full_name)
     user = crud.user.create(db, obj_in=user_in)
     return user
 
 
-@router.get("/{user_id}", response_model=User)
+@router.get("/{user_id}", response_model=schemas.User)
 def read_user_by_id(
     user_id: int,
-    current_user: DBUser = Depends(get_current_active_user),
-    db: Session = Depends(get_db),
-):
+    current_user: models.User = Depends(deps.get_current_active_user),
+    db: Session = Depends(deps.get_db),
+) -> Any:
     """
     Get a specific user by id.
     """
@@ -135,14 +132,14 @@ def read_user_by_id(
     return user
 
 
-@router.put("/{user_id}", response_model=User)
+@router.put("/{user_id}", response_model=schemas.User)
 def update_user(
     *,
-    db: Session = Depends(get_db),
+    db: Session = Depends(deps.get_db),
     user_id: int,
-    user_in: UserUpdate,
-    current_user: DBUser = Depends(get_current_active_superuser),
-):
+    user_in: schemas.UserUpdate,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
     """
     Update a user.
     """