✨ added structured api endpoints and done some refactoring

app/api/endpoints/items.py

@@ -0,0 +1,56 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+import crud
+import models
+from schemas.item import Item, ItemCreate, ItemUpdate
+from api import deps
+
+
+router = APIRouter()
+
+
+@router.get("/{item_id}", response_model=Item)
+async def get_item(
+    *,
+    db: Session = Depends(deps.get_db),
+    item_id: int
+):
+    return crud.item.get(db, item_id)
+
+
+@router.get("/", response_model=list[Item])
+async def get_items(
+    skip: int = 0,
+    limit: int = 100,
+    db: Session = Depends(deps.get_db)
+):
+    return crud.item.get_multi(db, skip=skip, limit=limit)
+
+
+@router.post("/", response_model=Item)
+async def create_item(
+    item: ItemCreate,
+    db: Session = Depends(deps.get_db)
+):
+    return crud.item.create(db=db, item=item)
+
+
+@router.put("/{item_id}", response_model=ItemUpdate)
+async def update_item(
+    item: ItemUpdate,
+    db: Session = Depends(deps.get_db)
+):
+    return crud.item.update(db=db, item=item)
+
+
+@router.delete("/{item_id}")
+async def delete_item(
+    *,
+    item: ItemUpdate,
+    db: Session = Depends(deps.get_db),
+    item_id: int
+):
+    return crud.remove(item_id)

app/api/endpoints/login.py

@@ -0,0 +1,96 @@
+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, models, schemas
+from app.api import deps
+from app.core import security
+from app.core.config import settings
+from app.core.security import get_password_hash
+from app.utils import (
+    generate_password_reset_token,
+    send_reset_password_email,
+    verify_password_reset_token,
+)
+
+router = APIRouter()
+
+
+@router.post("/login/access-token", response_model=schemas.Token)
+def login_access_token(
+    db: Session = Depends(deps.get_db), form_data: OAuth2PasswordRequestForm = Depends()
+) -> Any:
+    """
+    OAuth2 compatible token login, get an access token for future requests
+    """
+    user = crud.user.authenticate(
+        db, email=form_data.username, password=form_data.password
+    )
+    if not user:
+        raise HTTPException(status_code=400, detail="Incorrect email or password")
+    elif not crud.user.is_active(user):
+        raise HTTPException(status_code=400, detail="Inactive user")
+    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    return {
+        "access_token": security.create_access_token(
+            user.id, expires_delta=access_token_expires
+        ),
+        "token_type": "bearer",
+    }
+
+
+@router.post("/login/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=schemas.Msg)
+def recover_password(email: str, db: Session = Depends(deps.get_db)) -> Any:
+    """
+    Password Recovery
+    """
+    user = crud.user.get_by_email(db, email=email)
+
+    if not user:
+        raise HTTPException(
+            status_code=404,
+            detail="The user with this username does not exist in the system.",
+        )
+    password_reset_token = generate_password_reset_token(email=email)
+    send_reset_password_email(
+        email_to=user.email, email=email, token=password_reset_token
+    )
+    return {"msg": "Password recovery email sent"}
+
+
+@router.post("/reset-password/", response_model=schemas.Msg)
+def reset_password(
+    token: str = Body(...),
+    new_password: str = Body(...),
+    db: Session = Depends(deps.get_db),
+) -> Any:
+    """
+    Reset password
+    """
+    email = verify_password_reset_token(token)
+    if not email:
+        raise HTTPException(status_code=400, detail="Invalid token")
+    user = crud.user.get_by_email(db, email=email)
+    if not user:
+        raise HTTPException(
+            status_code=404,
+            detail="The user with this username does not exist in the system.",
+        )
+    elif not crud.user.is_active(user):
+        raise HTTPException(status_code=400, detail="Inactive user")
+    hashed_password = get_password_hash(new_password)
+    user.hashed_password = hashed_password
+    db.add(user)
+    db.commit()
+    return {"msg": "Password updated successfully"}

app/api/endpoints/users.py

@@ -0,0 +1,153 @@
+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, models, schemas
+from app.api import deps
+from app.core.config import settings
+from app.utils import send_new_account_email
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[schemas.User])
+def read_users(
+    db: Session = Depends(deps.get_db),
+    skip: int = 0,
+    limit: int = 100,
+    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=schemas.User)
+def create_user(
+    *,
+    db: Session = Depends(deps.get_db),
+    user_in: schemas.UserCreate,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
+    """
+    Create new user.
+    """
+    user = crud.user.get_by_email(db, email=user_in.email)
+    if user:
+        raise HTTPException(
+            status_code=400,
+            detail="The user with this username already exists in the system.",
+        )
+    user = crud.user.create(db, obj_in=user_in)
+    if settings.EMAILS_ENABLED and user_in.email:
+        send_new_account_email(
+            email_to=user_in.email, username=user_in.email, password=user_in.password
+        )
+    return user
+
+
+@router.put("/me", response_model=schemas.User)
+def update_user_me(
+    *,
+    db: Session = Depends(deps.get_db),
+    password: str = Body(None),
+    full_name: str = Body(None),
+    email: EmailStr = Body(None),
+    current_user: models.User = Depends(deps.get_current_active_user),
+) -> Any:
+    """
+    Update own user.
+    """
+    current_user_data = jsonable_encoder(current_user)
+    user_in = schemas.UserUpdate(**current_user_data)
+    if password is not None:
+        user_in.password = password
+    if full_name is not None:
+        user_in.full_name = full_name
+    if email is not None:
+        user_in.email = email
+    user = crud.user.update(db, db_obj=current_user, obj_in=user_in)
+    return user
+
+
+@router.get("/me", response_model=schemas.User)
+def read_user_me(
+    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=schemas.User)
+def create_user_open(
+    *,
+    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.
+    """
+    if not settings.USERS_OPEN_REGISTRATION:
+        raise HTTPException(
+            status_code=403,
+            detail="Open user registration is forbidden on this server",
+        )
+    user = crud.user.get_by_email(db, email=email)
+    if user:
+        raise HTTPException(
+            status_code=400,
+            detail="The user with this username already exists in the system",
+        )
+    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=schemas.User)
+def read_user_by_id(
+    user_id: int,
+    current_user: models.User = Depends(deps.get_current_active_user),
+    db: Session = Depends(deps.get_db),
+) -> Any:
+    """
+    Get a specific user by id.
+    """
+    user = crud.user.get(db, id=user_id)
+    if user == current_user:
+        return user
+    if not crud.user.is_superuser(current_user):
+        raise HTTPException(
+            status_code=400, detail="The user doesn't have enough privileges"
+        )
+    return user
+
+
+@router.put("/{user_id}", response_model=schemas.User)
+def update_user(
+    *,
+    db: Session = Depends(deps.get_db),
+    user_id: int,
+    user_in: schemas.UserUpdate,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
+    """
+    Update a user.
+    """
+    user = crud.user.get(db, id=user_id)
+    if not user:
+        raise HTTPException(
+            status_code=404,
+            detail="The user with this username does not exist in the system",
+        )
+    user = crud.user.update(db, db_obj=user, obj_in=user_in)
+    return user

app/api/endpoints/utils.py

@@ -0,0 +1,35 @@
+from typing import Any
+
+from fastapi import APIRouter, Depends
+from pydantic.networks import EmailStr
+
+from app import models, schemas
+from app.api import deps
+from app.core.celery_app import celery_app
+from app.utils import send_test_email
+
+router = APIRouter()
+
+
+@router.post("/test-celery/", response_model=schemas.Msg, status_code=201)
+def test_celery(
+    msg: schemas.Msg,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
+    """
+    Test Celery worker.
+    """
+    celery_app.send_task("app.worker.test_celery", args=[msg.msg])
+    return {"msg": "Word received"}
+
+
+@router.post("/test-email/", response_model=schemas.Msg, status_code=201)
+def test_email(
+    email_to: EmailStr,
+    current_user: models.User = Depends(deps.get_current_active_superuser),
+) -> Any:
+    """
+    Test emails.
+    """
+    send_test_email(email_to=email_to)
+    return {"msg": "Test email sent"}