Case Management Starter

app/applicants/models.py

@@ -0,0 +1,27 @@
+from sqlalchemy import Column, Date, Integer, String
+from sqlalchemy.orm import relationship
+
+from app.db import Base
+
+
+class DBApplicant(Base):
+    __tablename__ = "applicants"
+
+    id = Column(Integer, primary_key=True, index=True)
+    first_name = Column(String, nullable=False)
+    last_name = Column(String, nullable=False)
+    middle_name = Column(String, nullable=True)
+    gender = Column(String, nullable=False)
+    date_of_birth = Column(Date, nullable=False)
+    ssn = Column(String, nullable=False, unique=True)
+    email = Column(String, nullable=True)
+    home_phone = Column(String, nullable=True)
+    mobile_phone = Column(String, nullable=True)
+    address = Column(String, nullable=True)
+    city = Column(String, nullable=True)
+    state = Column(String, nullable=True)
+    zip = Column(String, nullable=True)
+    country = Column(String, nullable=False)
+    created_at = Column(Date, nullable=False)
+    updated_at = Column(Date, nullable=False)
+    case = relationship("DBCase", back_populates="applicant")

app/applicants/router.py

@@ -0,0 +1,53 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from starlette import status
+
+import app.applicants.services as service
+from app.applicants.schemas import Applicant, ApplicantPayload
+from app.db import get_db
+
+router = APIRouter(
+    prefix="/api",
+    tags=["Applicants"],
+    responses={404: {"description": "Endpoint not found"}},
+)
+
+# Database dependency injection session
+db_session = Annotated[Session, Depends(get_db)]
+
+
+@router.get(
+    "/applicants", status_code=status.HTTP_200_OK, response_model=ApplicantPayload
+)
+async def get_applicants(db: db_session, page_number: int = 0, page_size: int = 100):
+    return service.get_items(db, page_number, page_size)
+
+
+@router.get(
+    "/applicants/{id}", status_code=status.HTTP_200_OK, response_model=Applicant
+)
+async def get_applicant(id: int, db: db_session):
+    return service.get_item(db, id)
+
+
+@router.put(
+    "/applicants/{id}", status_code=status.HTTP_200_OK, response_model=Applicant
+)
+async def update_applicant(id: int, applicant: Applicant, db: db_session):
+    db_applicant = service.update_item(db, id, applicant)
+    return db_applicant
+
+
+@router.post(
+    "/applicants", status_code=status.HTTP_201_CREATED, response_model=Applicant
+)
+async def create_applicant(applicant: Applicant, db: db_session):
+    db_applicant = service.create_item(db, applicant)
+    return db_applicant
+
+
+@router.delete("/applicants/{id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_applicant(id: int, db: db_session):
+    service.delete_item(db, id)

app/applicants/schemas.py

@@ -0,0 +1,33 @@
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+# Pydantic Models
+class Applicant(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int | None = None
+    first_name: str
+    last_name: str
+    middle_name: str | None = None
+    gender: str
+    date_of_birth: date
+    ssn: str
+    email: str | None = None
+    home_phone: str | None = None
+    mobile_phone: str | None = None
+    address: str | None = None
+    city: str | None = None
+    state: str | None = None
+    zip: str | None = None
+    country: str = "USA"
+    created_at: datetime
+    updated_at: datetime
+
+
+class ApplicantPayload(BaseModel):
+    items: list[Applicant]
+    item_count: int = 0
+    page_count: int = 0
+    prev_page: int | None = None
+    next_page: int | None = None

app/applicants/services.py

@@ -0,0 +1,71 @@
+from datetime import datetime
+
+from fastapi import HTTPException
+from sqlalchemy.orm import Session
+
+from app.applicants.models import DBApplicant
+from app.applicants.schemas import Applicant
+from app.utils import get_next_page, get_page_count, get_prev_page
+
+
+def get_items(db: Session, page_number: int, page_size: int):
+    item_count = db.query(DBApplicant).count()
+    items = db.query(DBApplicant).limit(page_size).offset(page_number * page_size).all()
+
+    return {
+        "items": items,
+        "item_count": item_count,
+        "page_count": get_page_count(item_count, page_size),
+        "prev_page": get_prev_page(page_number),
+        "next_page": get_next_page(item_count, page_number, page_size),
+    }
+
+
+def get_item(db: Session, applicant_id: int):
+    return db.query(DBApplicant).where(DBApplicant.id == applicant_id).first()
+
+
+def update_item(db: Session, id: int, applicant: Applicant):
+    db_applicant = db.query(DBApplicant).filter(DBApplicant.id == id).first()
+    if db_applicant is None:
+        raise HTTPException(status_code=404, detail="Applicant not founds")
+
+    db_applicant.first_name = applicant.first_name
+    db_applicant.last_name = applicant.last_name
+    db_applicant.middle_name = applicant.middle_name
+    db_applicant.gender = applicant.gender
+    db_applicant.date_of_birth = applicant.date_of_birth
+    db_applicant.ssn = applicant.ssn
+    db_applicant.email = applicant.email
+    db_applicant.home_phone = applicant.home_phone
+    db_applicant.mobile_phone = applicant.mobile_phone
+    db_applicant.address = applicant.address
+    db_applicant.city = applicant.city
+    db_applicant.state = applicant.state
+    db_applicant.zip = applicant.zip
+    db_applicant.country = applicant.country
+    db_applicant.date_of_birth = applicant.date_of_birth
+    db_applicant.updated_at = datetime.now()
+    db.add(db_applicant)
+    db.commit()
+    db.refresh(db_applicant)
+
+    return db_applicant
+
+
+def create_item(db: Session, applicant: Applicant):
+    db_applicant = DBApplicant(**applicant.model_dump())
+    db.add(db_applicant)
+    db.commit()
+    db.refresh(db_applicant)
+
+    return db_applicant
+
+
+def delete_item(db: Session, id: int):
+    db_applicant = db.query(DBApplicant).filter(DBApplicant.id == id).first()
+    if db_applicant is None:
+        raise HTTPException(status_code=404, detail="Applicant not founds")
+
+    db.query(DBApplicant).filter(DBApplicant.id == id).delete()
+    db.commit()

app/cases/models.py

@@ -0,0 +1,19 @@
+from sqlalchemy import Column, Date, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from app.db import Base
+
+
+# SQLAlchemy Model
+class DBCase(Base):
+    __tablename__ = "cases"
+
+    id = Column(Integer, primary_key=True, index=True)
+    status = Column(String, nullable=False)
+    assigned_to = Column(String, nullable=True)
+    created_at = Column(Date, nullable=False)
+    updated_at = Column(Date, nullable=False)
+    applicant_id = Column(
+        Integer, ForeignKey("applicants.id", ondelete="CASCADE"), nullable=False
+    )
+    applicant = relationship("DBApplicant", back_populates="case")

app/cases/router.py

@@ -0,0 +1,47 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from starlette import status
+
+import app.cases.services as service
+from app.cases.schemas import Case, CasePayload, CaseWithApplicant
+from app.db import get_db
+
+router = APIRouter(
+    prefix="/api",
+    tags=["Cases"],
+    responses={404: {"description": "Endpoint not found"}},
+)
+
+# Database dependency injection session
+db_session = Annotated[Session, Depends(get_db)]
+
+
+@router.get("/cases", status_code=status.HTTP_200_OK, response_model=CasePayload)
+async def get_cases(db: db_session, page_number: int = 0, page_size: int = 100):
+    return service.get_items(db, page_number, page_size)
+
+
+@router.get(
+    "/cases/{id}", status_code=status.HTTP_200_OK, response_model=CaseWithApplicant
+)
+async def get_case(id: int, db: db_session):
+    return service.get_item(db, id)
+
+
+@router.put("/cases/{id}", status_code=status.HTTP_200_OK, response_model=Case)
+async def update_case(id: int, case: Case, db: db_session):
+    db_case = service.update_item(db, id, case)
+    return db_case
+
+
+@router.post("/cases", status_code=status.HTTP_201_CREATED, response_model=Case)
+async def create_case(case: Case, db: db_session):
+    db_case = service.create_item(db, case)
+    return db_case
+
+
+@router.delete("/cases/{id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_case(id: int, db: db_session):
+    service.delete_item(db, id)

app/cases/schemas.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from app.applicants.schemas import Applicant
+
+
+# Pydantic Models
+class Case(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int | None = None
+    status: str
+    assigned_to: str | None = None
+    created_at: datetime
+    updated_at: datetime
+    applicant_id: int | None = None
+
+
+class CaseWithApplicant(BaseModel):
+    id: int
+    status: str
+    assigned_to: str | None = None
+    created_at: datetime
+    updated_at: datetime
+    applicant: Applicant | None = None
+
+
+class CasePayload(BaseModel):
+    items: list[CaseWithApplicant]
+    item_count: int = 0
+    page_count: int = 0
+    prev_page: int | None = None
+    next_page: int | None = None

app/cases/services.py

@@ -0,0 +1,98 @@
+from datetime import datetime
+
+from fastapi import HTTPException
+from sqlalchemy.orm import Session, joinedload
+
+from app.cases.models import DBCase
+from app.cases.schemas import Case
+from app.utils import get_next_page, get_page_count, get_prev_page
+
+
+def get_items(db: Session, page_number: int, page_size: int):
+    item_count = db.query(DBCase).count()
+    items = db.query(DBCase).limit(page_size).offset(page_number * page_size).all()
+
+    return {
+        "items": items,
+        "item_count": item_count,
+        "page_count": get_page_count(item_count, page_size),
+        "prev_page": get_prev_page(page_number),
+        "next_page": get_next_page(item_count, page_number, page_size),
+    }
+
+
+def get_item(db: Session, case_id: int):
+    case = (
+        db.query(DBCase)
+        .options(joinedload(DBCase.applicant))
+        .where(DBCase.id == case_id)
+        .first()
+    )
+
+    if case is None:
+        raise HTTPException(status_code=404, detail="Case not found")
+
+    # Handle case where applicant might be None
+    applicant_data = None
+    if case.applicant:
+        applicant_data = {
+            "id": case.applicant.id,
+            "first_name": case.applicant.first_name,
+            "last_name": case.applicant.last_name,
+            "middle_name": case.applicant.middle_name,
+            "email": case.applicant.email,
+            "gender": case.applicant.gender,
+            "date_of_birth": case.applicant.date_of_birth,
+            "ssn": case.applicant.ssn,
+            "home_phone": case.applicant.home_phone,
+            "mobile_phone": case.applicant.mobile_phone,
+            "address": case.applicant.address,
+            "city": case.applicant.city,
+            "state": case.applicant.state,
+            "zip": case.applicant.zip,
+            "country": case.applicant.country,
+            "created_at": case.applicant.created_at,
+            "updated_at": case.applicant.updated_at,
+        }
+
+    return {
+        "id": case.id,
+        "status": case.status,
+        "applicant": applicant_data,
+        "assigned_to": case.assigned_to,
+        "created_at": case.created_at,
+        "updated_at": case.updated_at,
+    }
+
+
+def update_item(db: Session, id: int, case: Case):
+    db_case = db.query(DBCase).filter(DBCase.id == id).first()
+    if db_case is None:
+        raise HTTPException(status_code=404, detail="Case not founds")
+
+    db_case.status = case.status
+    db_case.assigned_to = case.assigned_to
+    db_case.updated_at = datetime.now()
+    db.add(db_case)
+    db.commit()
+    db.refresh(db_case)
+
+    return db_case
+
+
+def create_item(db: Session, case: Case):
+    db_case = DBCase(**case.model_dump())
+    db.add(db_case)
+    db.commit()
+    db.refresh(db_case)
+
+    return db_case
+
+
+def delete_item(db: Session, id: int):
+    db_case = db.query(DBCase).filter(DBCase.id == id).first()
+    if db_case is None:
+        raise HTTPException(status_code=404, detail="Case not founds")
+
+    db.query(DBCase).filter(DBCase.id == id).delete()
+    db.commit()