# ems_app.py
from fastapi import FastAPI, HTTPException, status, Query
from pydantic import BaseModel, Field
app = FastAPI(
title="Employee Management System (In-Memory)",
description="Foundational CRUD API for managing employees.",
version="1.0.0"
)
# ----------------- PYDANTIC SCHEMAS -----------------
class EmployeeBase(BaseModel):
name: str = Field(..., min_length=2, max_length=50, examples=["Jane Doe"])
department: str = Field(..., examples=["Engineering"])
role: str = Field(..., examples=["Software Engineer"])
class EmployeeCreate(EmployeeBase):
pass
class EmployeeUpdate(BaseModel):
name: str | None = Field(None, min_length=2, max_length=50)
department: str | None = Field(None)
role: str | None = Field(None)
class EmployeeOut(EmployeeBase):
id: int
# ----------------- MOCK DATA STATE -----------------
EMPLOYEES: dict[int, dict] = {
1: {"id": 1, "name": "Alice Smith", "department": "Engineering", "role": "Backend Developer"},
2: {"id": 2, "name": "Bob Jones", "department": "Product", "role": "Product Manager"}
}
# ----------------- ENDPOINTS (CRUD) -----------------
# 1. READ ALL (Retrieve a list of employees)
@app.get("/employees", response_model=list[EmployeeOut])
def list_employees(
department: str | None = Query(None, description="Filter employees by department name"),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100)
):
results = list(EMPLOYEES.values())
if department:
results = [emp for emp in results if emp["department"].lower() == department.lower()]
return results[skip : skip + limit]
# 2. READ ONE (Retrieve a single employee by ID)
@app.get("/employees/{employee_id}", response_model=EmployeeOut)
def get_employee(employee_id: int):
employee = EMPLOYEES.get(employee_id)
if not employee:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Employee with ID {employee_id} not found."
)
return employee
# 3. CREATE (Onboard a new employee)
@app.post("/employees", response_model=EmployeeOut, status_code=status.HTTP_201_CREATED)
def create_employee(employee: EmployeeCreate):
new_id = max(EMPLOYEES.keys()) + 1 if EMPLOYEES else 1
new_employee = {"id": new_id, **employee.model_dump()}
EMPLOYEES[new_id] = new_employee
return new_employee
# 4. UPDATE (Modify details of an existing employee)
@app.put("/employees/{employee_id}", response_model=EmployeeOut)
def update_employee(employee_id: int, employee_update: EmployeeUpdate):
if employee_id not in EMPLOYEES:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Employee with ID {employee_id} not found."
)
stored_employee = EMPLOYEES[employee_id]
# Update only the fields that were provided in the request
update_data = employee_update.model_dump(exclude_unset=True)
for key, value in update_data.items():
stored_employee[key] = value
EMPLOYEES[employee_id] = stored_employee
return stored_employee
# 5. DELETE (Offboard/Remove an employee)
@app.delete("/employees/{employee_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_employee(employee_id: int):
if employee_id not in EMPLOYEES:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Employee with ID {employee_id} not found."
)
del EMPLOYEES[employee_id]
return # HTTP 204 requires no return body
