Full refactor/ add unit tests

Author: Eric Sales De Andrade

.env

@@ -2,7 +2,7 @@
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
-SQLITE_DATABASE_URL = "sqlite:///./note.db"
+SQLITE_DATABASE_URL = "sqlite:///./user.db"
 
 engine = create_engine(
  SQLITE_DATABASE_URL, echo=True, connect_args={"check_same_thread": False}

Makefile

@@ -1,7 +1,7 @@
 from app import models, user
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from .database import engine
+from app.database import engine
 
 models.Base.metadata.create_all(bind=engine)
 
@@ -20,7 +20,7 @@
 )
 
 
-app.include_router(user.router, tags=['Users'], prefix='/api/users')
+app.include_router(user.router, tags=["Users"], prefix="/api/users")
 
 
 @app.get("/api/healthchecker")

app/database.py

@@ -1,4 +1,4 @@
-from database import Base
+from app.database import Base
 from sqlalchemy import TIMESTAMP, Column, String, Boolean
 from sqlalchemy.sql import func
 from fastapi_utils.guid_type import GUID, GUID_DEFAULT_SQLITE
@@ -11,7 +11,7 @@ class User(Base):
  last_name = Column(String, nullable=False)
  address = Column(String, nullable=True)
  activated = Column(Boolean, nullable=False, default=True)
- createdAt = Column(TIMESTAMP(timezone=True),
-  nullable=False, server_default=func.now())
- updatedAt = Column(TIMESTAMP(timezone=True),
-   default=None, onupdate=func.now())
+ createdAt = Column(
+ TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
+ )
+ updatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())

app/main.py

@@ -1,21 +1,28 @@
-import schemas, models
+import app.schemas as schemas, app.models as models
 from sqlalchemy.orm import Session
-from fastapi import Depends, HTTPException, status, APIRouter, Response
-from database import get_db
+from fastapi import Depends, HTTPException, status, APIRouter
+from app.database import get_db
 
 router = APIRouter()
 
 
 @router.get("/", status_code=status.HTTP_200_OK)
-def get_users(db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ''):
+def get_users(
+ db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""
+):
  skip = (page - 1) * limit
 
- users = db.query(models.User).filter(
- models.User.title.contains(search)).limit(limit).offset(skip).all()
+ users = (
+ db.query(models.User)
+ .filter(models.User.first_name.contains(search))
+ .limit(limit)
+ .offset(skip)
+ .all()
+ )
  return {"Status": "Success", "Results": len(users), "Users": users}
 
 
-@router.post('/', status_code=status.HTTP_201_CREATED)
+@router.post("/", status_code=status.HTTP_201_CREATED)
 def create_user(payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):
  new_user = models.User(**payload.dict())
  db.add(new_user)
@@ -25,37 +32,46 @@ def create_user(payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):
 
 
 @router.patch("/{userId}", status_code=status.HTTP_202_ACCEPTED)
-def update_user(noteId: str, payload: schemas.UserBaseSchema, db: Session = Depends(get_db)):
- user_query = db.query(models.User).filter(models.User.id == noteId)
+def update_user(
+ userId: str, payload: schemas.UserBaseSchema, db: Session = Depends(get_db)
+):
+ user_query = db.query(models.User).filter(models.User.id == userId)
  db_user = user_query.first()
 
  if not db_user:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
- detail=f'No note with this id: {noteId} found')
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"No User with this id: {userId} found",
+ )
  update_data = payload.dict(exclude_unset=True)
- user_query.filter(models.User.id == noteId).update(update_data,
- synchronize_session=False)
+ user_query.filter(models.User.id == userId).update(
+ update_data, synchronize_session=False
+ )
  db.commit()
  db.refresh(db_user)
  return {"Status": "Success", "User": db_user}
 
 
 @router.get("/{userId}", status_code=status.HTTP_200_OK)
-def get_post(userId: str, db: Session = Depends(get_db)):
+def get_user(userId: str, db: Session = Depends(get_db)):
  user = db.query(models.User).filter(models.User.id == userId).first()
  if not user:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
- detail=f"No User with this id: {userId} found")
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"No User with this id: `{userId}` found",
+ )
  return {"Status": "Success", "User": user}
 
 
 @router.delete("/{userId}", status_code=status.HTTP_200_OK)
-def delete_post(userId: str, db: Session = Depends(get_db)):
+def delete_user(userId: str, db: Session = Depends(get_db)):
  user_query = db.query(models.User).filter(models.User.id == userId)
  user = user_query.first()
  if not user:
- raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
- detail=f'No user with this id: {userId} found')
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"No user with this id: {userId} found",
+ )
  user_query.delete(synchronize_session=False)
  db.commit()
  return {"Status": "Success", "Message": "User deleted successfully"}

app/models.py

@@ -0,0 +1,4 @@
+[pytest]
+log_cli=true
+log_level=INFO
+addopts = -p no:warnings

app/user.py

@@ -1,37 +1,39 @@
 anyio==3.6.2
-autopep8==1.7.1
-certifi==2022.9.24
-charset-normalizer==2.1.1
+autopep8==2.0.2
+certifi==2022.12.7
+charset-normalizer==3.1.0
 click==8.1.3
 colorama==0.4.6
-dnspython==2.2.1
-email-validator==1.3.0
-fastapi==0.85.1
+dnspython==2.3.0
+email-validator==1.3.1
+fastapi==0.94.1
 fastapi-utils==0.2.1
-greenlet==1.1.3.post0
+greenlet==2.0.2
 h11==0.14.0
 httptools==0.5.0
 idna==3.4
 itsdangerous==2.1.2
 Jinja2==3.1.2
-MarkupSafe==2.1.1
-orjson==3.8.1
+MarkupSafe==2.1.2
+orjson==3.8.7
 psycopg2==2.9.5
-pycodestyle==2.9.1
-pydantic==1.10.2
-python-dotenv==0.21.0
-python-multipart==0.0.5
+pycodestyle==2.10.0
+pydantic==1.10.6
+python-dotenv==1.0.0
+python-multipart==0.0.6
 PyYAML==6.0
-requests==2.28.1
+requests==2.28.2
 six==1.16.0
 sniffio==1.3.0
-SQLAlchemy==1.4.42
-starlette==0.20.4
+SQLAlchemy==1.4.46
+starlette==0.26.1
 tomli==2.0.1
-typing_extensions==4.4.0
-ujson==5.5.0
-urllib3==1.26.12
-uvicorn==0.18.3
-watchfiles==0.18.0
+typing_extensions==4.5.0
+ujson==5.7.0
+urllib3==1.26.15
+uvicorn==0.21.1
+watchfiles==0.18.1
 websockets==10.4
 pytest==7.2.2
+httpx==0.23.3
+black==23.1.0

docker-compose.yml

@@ -1,35 +1,100 @@
-import uuid
 from fastapi.testclient import TestClient
-
 from app.main import app
+import uuid
 
 client = TestClient(app)
-note_id = str(uuid.uuid4())
+
+user_id = str(uuid.uuid4())
 
 
 def test_root():
  response = client.get("/api/healthchecker")
  assert response.status_code == 200
- assert response.json() == {"message": "Welcome to FastAPI with SQLAlchemy"}
+ assert response.json() == {"message": "The API is LIVE!!"}
 
 
-def test_get_notes():
- response = client.get("/api/notes/")
+def test_create_user():
+ sample_payload = {
+ "id": user_id,
+ "first_name": "PLACEHOLDER",
+ "last_name": "PLACEHOLDER",
+ "address": "PLACEHOLDER",
+ "activated": False,
+ "createdAt": "2023-03-17T00:04:32",
+ }
+ response = client.post("/api/users/", json=sample_payload)
+ assert response.status_code == 201
+ assert response.json() == {
+ "Status": "Success",
+ "User": {
+ "first_name": "PLACEHOLDER",
+ "last_name": "PLACEHOLDER",
+ "activated": False,
+ "createdAt": "2023-03-17T00:04:32",
+ "id": user_id,
+ "address": "PLACEHOLDER",
+ "updatedAt": None,
+ },
+ }
+
+
+def test_get_user():
+ response = client.get(f"/api/users/{user_id}")
  assert response.status_code == 200
- assert response.json() == {'status': 'success', 'results': 0, 'notes': []}
+ assert response.json() == {
+ "Status": "Success",
+ "User": {
+ "first_name": "PLACEHOLDER",
+ "last_name": "PLACEHOLDER",
+ "activated": False,
+ "createdAt": "2023-03-17T00:04:32",
+ "address": "PLACEHOLDER",
+ "id": user_id,
+ "updatedAt": None,
+ },
+ }
 
 
-def test_create_note():
+def test_update_user():
  sample_payload = {
- "id": note_id,
- "title": "string",
- "content": "string",
- "category": "string",
- "published": False,
- "createdAt": "2023-03-14T09:11:48.620Z",
- "updatedAt": "2023-03-14T09:11:48.620Z"
- }
- response = client.post("/api/notes/", json=sample_payload)
- assert response.status_code == 201
- assert response.json() == {'status': 'success', 'note': {'published': False, 'title': 'string', 'content': 'string', 'createdAt': '2023-03-14T09:11:48.620000', 'id': note_id, 'category': 'string', 'updatedAt': '2023-03-14T09:11:48.620000'}}
+ "id": user_id,
+ "first_name": "PLACEHOLDER2",
+ "last_name": "PLACEHOLDER2",
+ "address": "PLACEHOLDER2",
+ "activated": True,
+ "createdAt": "2023-03-17T00:04:32",
+ "updatedAt": "2023-03-17T00:06:32",
+ }
+ response = client.patch(f"/api/users/{user_id}", json=sample_payload)
+ assert response.status_code == 202
+ assert response.json() == {
+ "Status": "Success",
+ "User": {
+ "first_name": "PLACEHOLDER2",
+ "last_name": "PLACEHOLDER2",
+ "activated": True,
+ "createdAt": "2023-03-17T00:04:32",
+ "id": user_id,
+ "address": "PLACEHOLDER2",
+ "updatedAt": "2023-03-17T00:06:32",
+ },
+ }
+
+
+def test_delete_user():
+ response = client.delete(f"/api/users/{user_id}")
+ assert response.status_code == 200
+ assert response.json() == {
+ "Status": "Success",
+ "Message": "User deleted successfully",
+ }
+
 
+def test_get_user_not_found():
+ response = client.get(
+ f"/api/users/16303002-876a-4f39-ad16-e715f151bab3"
+ ) # GUID not in DB
+ assert response.status_code == 404
+ assert response.json() == {
+ "detail": "No User with this id: `16303002-876a-4f39-ad16-e715f151bab3` found"
+ }