Quality Defect Workflow API
Build a FastAPI backend for a controlled vehicle quality defect workflow.
Introduction
30 Second Summary
On a busy vehicle line, one rushed status change can make an unresolved defect look finished. That weak record makes it harder to prove what the team contained or corrected.
In this project, you will build a local FastAPI backend that records vehicle inspections with linked quality defects. The API enforces containment before closure while preserving a timestamped audit trail.
What You'll Build
Your final demo rejects an unsafe closure request before returning the correctly closed defect with its full audit history.
By the end of this project, you'll have:
- A working local API where you create an inspection before reporting a linked defect against its generated identifier.
- A controlled status workflow that visibly rejects any attempt to close an uncontained defect.
- A timestamped audit trail that shows who changed each defect at every accepted transition.
- Secret Mission: Add a filterable defect queue that narrows records by station, severity, or workflow status.
Are there any prerequisites?
You'll need a Linux computer with Python 3.10 or later plus basic FastAPI familiarity. The walkthrough covers the isolated virtual environment plus pinned packages.
Before We Start
This step defines the factory rule your API will protect before any hands-on work begins. That rule gives every later decision a clear purpose.
Set Up the Quality Workflow API
Your quality workflow needs a stable local home before you add inspections or defects. An isolated Python environment prevents package conflicts from changing how the API behaves.
In this step, you will create the project with pinned versions of FastAPI and Pydantic. You will finish with a health-check route running on your computer.
In this step, get ready to:
- Create the project directory and verify Python 3.10 or later.
- Create an isolated virtual environment and install the pinned dependencies.
- Run a starter API with local interactive documentation.
Create the project and verify Python
FastAPI 0.143.0 requires Python 3.10 or later. Checking the interpreter now protects the dependency installation that follows.
- Use your Linux application launcher to open a terminal.
- Move to your Desktop by running this command:
cd ~/Desktop
What does this command do?
The cd command changes the terminal's current location. Starting from your Desktop makes the project directory easy to find.
- Create the project directory and check its Python version by running these commands:
mkdir quality-defect-workflow
cd quality-defect-workflow
python --version
What do these commands do?
- The first command creates the quality-defect-workflow directory on your Desktop.
- The second command moves the terminal into that directory.
- The final command prints the Python version used by this project.
Compare the reported version with the project's minimum before you continue.
✔️ I see a supported version
Your terminal reports Python 3.10 or later. That interpreter can install both pinned packages.
Good start. Your project directory now has the runtime it needs.
ⓧ I see an older version
Your current interpreter cannot run the pinned package versions. Install Python 3.10 or later before creating the virtual environment.
- Follow the official Python guidance for Linux to install a supported interpreter through your Linux distribution or from source.
- Configure your shell so the python command uses the supported interpreter.
- Repeat the version check shown above.
ⓧ Command not found
Your shell cannot find a Python interpreter through the python command. Install Python 3.10 or later before continuing.
- Follow the official Python guidance for Linux to install a supported interpreter.
- Configure your shell so the python command finds that installation.
- Repeat the version check shown above.
Isolate and install the dependencies
A virtual environment gives this project its own package directory. The pinned requirements then make every installation use the same FastAPI and Pydantic releases.
- Create the virtual environment inside quality-defect-workflow by running this command:
python -m venv .venv
What does this command do?
Python's venv module creates an isolated environment in .venv. Packages installed after activation stay inside this project.
The command returns to the prompt when the .venv directory is ready.
Virtual environment creation failed?
Confirm that the terminal is still inside quality-defect-workflow. Your Linux installation may also require its virtual-environment support package.
Help me fix the virtual environment creation error.
- Activate the virtual environment by running this command:
source .venv/bin/activate
What does activation change?
Activation points the terminal at the environment's Python interpreter and package installer. Future package commands now affect .venv.
Your prompt should now begin with (.venv). That visible prefix confirms the environment is active.
Missing the activation prefix?
Check that .venv exists inside quality-defect-workflow. Run the activation command from that directory.
Help me activate my virtual environment.
- Open the current project directory in Visual Studio Code by running this command:
code .
What does this command do?
The code . command opens the terminal's current directory as a Visual Studio Code workspace. The editor can now show every project file in one Explorer sidebar.
You should see a Visual Studio Code window with quality-defect-workflow in the Explorer sidebar.
Code command not found?
Use your Linux application launcher to start Visual Studio Code. Open the quality-defect-workflow directory from your Desktop.
Help me open my project in Visual Studio Code.
- Select the new-file icon at the top of the Visual Studio Code Explorer sidebar.
- Enter requirements.txt as the file name.
- Add the pinned dependencies by pasting this content:
fastapi[standard]==0.143.0
pydantic==2.14.0
Why pin these versions?
The first line installs FastAPI 0.143.0 with its standard development tools. The second line installs Pydantic 2.14.0 for typed request and response data.
Exact pins keep the project reproducible across different computers.
- Press Ctrl+S (Linux/Windows) or Cmd+S (macOS) to save requirements.txt.
The installation can take a minute while the packages download. A short pause in the terminal is expected.
- Switch back to the activated terminal from earlier.
- Install the pinned dependencies by running this command:
python -m pip install -r requirements.txt
What does this command do?
Python runs pip from the active virtual environment. The -r option tells it to install every requirement listed in requirements.txt.
You should see package download and installation output complete without an error. Your terminal prompt should still begin with (.venv).
Dependency installation failed?
Confirm that requirements.txt contains both pinned lines exactly. Confirm that the terminal prompt still starts with (.venv).
Help me fix the dependency installation.
Build and run the starter API
The first route acts as a health check. It gives you immediate proof that FastAPI can import the project and serve a response before the workflow logic grows.
- Select the new-file icon at the top of the Visual Studio Code Explorer sidebar.
- Enter main.py as the file name.
- Define the starter application by pasting this code:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def health_check() -> dict[str, str]:
return {"status": "quality workflow API is running"}
What does this code do?
- The import makes the FastAPI application class available to this file.
- The app variable creates the API instance that the development server detects.
- The @app.get("/") decorator connects a request for the root path to health_check().
- The returned dictionary becomes the API's JSON health response.
- Press Ctrl+S (Linux/Windows) or Cmd+S (macOS) to save main.py.
Before you run the server, do you expect FastAPI to expose only the health route or interactive documentation too?
- Switch back to the activated terminal from earlier.
- Start the development server by running this command:
fastapi dev
What does this command do?
The FastAPI development command reads main.py and detects the app instance. It starts a local server with automatic reload enabled.
Automatic reload restarts the server whenever you save a Python change.
You should see the terminal report a server at http://127.0.0.1:8000. It also reports interactive documentation at http://127.0.0.1:8000/docs.
- Open http://127.0.0.1:8000 in a browser if one is available.
You should see a JSON response containing quality workflow API is running.
- Open http://127.0.0.1:8000/docs in a browser if one is available.
You should see the interactive API documentation with the root health-check route listed. The terminal remains the primary success signal when no browser is available.
Keep the server running
Leave this terminal running for the next step. The server reloads as you add the inspection and defect models.
Server did not start?
Confirm that the prompt starts with (.venv). Check that main.py is saved inside quality-defect-workflow.
Help me start the FastAPI development server.
✔️ Awesome, I've got everything!
Your pinned dependencies are saved. Your starter API is running from the activated virtual environment.
ⓧ I'd like to double check the full code
Compare your requirements.txt file with this complete version.
fastapi[standard]==0.143.0
pydantic==2.14.0
Compare your main.py file with this complete version.
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def health_check() -> dict[str, str]:
return {"status": "quality workflow API is running"}
Your isolated FastAPI service is live with pinned dependencies. Next, you will model inspections and linked defects.
Record Inspections and Defects
Your starter FastAPI service is live. It can confirm that the quality workflow API is running.
The API now needs to capture the inspection context behind each quality problem. This creates the visible foundation for the workflow rules you add later.
Typed Pydantic models define the shape of each request. A UUID links every defect to its inspection. Each new defect also starts an audit trail that records its first workflow event.
In this step, get ready to:
- Define typed production stations, severity levels, workflow statuses, inspections, defects, and audit events.
- Store inspections and defects in memory with UUID keys.
- Create linked records and retrieve a defect through the interactive documentation.
Define the quality data contract
An enum limits a field to a known set of values. The request and response models use those values to keep inspection and defect data consistent.
- In main.py, replace the current import line with these imports:
from datetime import datetime, timezone
from enum import Enum
from uuid import UUID, uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
What do these imports provide?
- The datetime imports create timezone-aware event timestamps.
- The enum import provides controlled string values for the production data.
- The UUID imports provide record identifiers.
- The FastAPI and Pydantic imports support HTTP errors plus typed data models.
- Save main.py.
The running server reloads the file without a traceback. This confirms that every new import is available.
Does the server stop reloading?
Check that each import matches the snippet exactly. A missing comma or module name prevents Python from loading main.py.
Return to the running terminal after fixing the import. Help me troubleshoot the imports in my FastAPI application.
- In main.py, find app = FastAPI().
- Add these enum definitions below that line:
class ProductionStation(str, Enum):
BODY_SHOP = "body_shop"
PAINT_SHOP = "paint_shop"
FINAL_ASSEMBLY = "final_assembly"
class Severity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
class DefectStatus(str, Enum):
OPEN = "open"
CONTAINED = "contained"
CLOSED = "closed"
What do these enums control?
- ProductionStation identifies where the inspection happened.
- Severity describes the urgency of a reported defect.
- DefectStatus supplies the three states used by the quality workflow.
- Save main.py.
The server reloads successfully. Your API now has a controlled vocabulary for plant records.
Seeing an enum syntax error?
Check that each class inherits from str and Enum. Confirm that every enum value has matching quotation marks.
Help me find the syntax error in my enum definitions.
- Add the inspection request model below DefectStatus by copying this code:
class InspectionCreate(BaseModel):
vehicle_id: str
station: ProductionStation
inspector: str
What does this model describe?
- InspectionCreate defines the information that a caller supplies.
- The station field accepts only a ProductionStation value.
- Save main.py.
The server reloads successfully. The inspection request now has a typed structure.
Does InspectionCreate fail to load?
Confirm that InspectionCreate appears below the enum classes. Check that every field is indented with four spaces.
Help me repair my InspectionCreate model.
- Add the stored inspection model below InspectionCreate by copying this code:
class InspectionRecord(InspectionCreate):
id: UUID
inspected_at: datetime
How does the stored record grow?
- InspectionRecord inherits every request field from InspectionCreate.
- The API adds an identifier plus the inspection time before storing the record.
- Save main.py.
The server reloads successfully. Inspection responses can now include generated metadata.
Is InspectionRecord undefined?
Place InspectionRecord directly below InspectionCreate. Confirm that the parent class name uses the same capitalization.
Help me fix the relationship between my inspection models.
- Add the defect request model below InspectionRecord by copying this code:
class DefectCreate(BaseModel):
inspection_id: UUID
title: str
description: str
severity: Severity
reported_by: str
What links a defect to its context?
- The inspection_id field points to the inspection that produced the defect.
- The remaining fields capture the quality problem plus its reporter.
- Save main.py.
The server reloads successfully. Defect requests now carry a link back to their inspection.
Does DefectCreate trigger an error?
Check that inspection_id uses UUID. Confirm that severity uses the Severity enum defined earlier.
Help me debug my DefectCreate model.
- Add the audit event model below DefectCreate by copying this code:
class AuditEvent(BaseModel):
timestamp: datetime
actor: str
action: str
from_status: DefectStatus | None = None
to_status: DefectStatus
What does an audit event preserve?
- The timestamp records when the event happened.
- The actor and action fields identify who changed the record plus what happened.
- The status fields preserve the workflow movement represented by the event.
- Save main.py.
The server reloads successfully. Each defect can now carry a typed history entry.
Seeing an error near the status fields?
Confirm that from_status includes | None. Check that its default value is None.
Help me fix the optional status field in AuditEvent.
- Add the stored defect model and in-memory collections below AuditEvent by copying this code:
class DefectRecord(DefectCreate):
id: UUID
status: DefectStatus
containment_action: str | None = None
corrective_action: str | None = None
audit_trail: list[AuditEvent]
inspections: dict[UUID, InspectionRecord] = {}
defects: dict[UUID, DefectRecord] = {}
Where do the records live?
- DefectRecord extends the incoming defect with workflow state plus audit history.
- The inspections dictionary stores InspectionRecord values under UUID keys.
- The defects dictionary stores DefectRecord values under UUID keys.
- Save main.py.
The server reloads successfully. The complete quality data contract is now ready for API operations.
Does the server fail on DefectRecord?
Confirm that AuditEvent appears before DefectRecord. Check that both dictionaries appear before the health check operation.
Help me debug the stored defect model and dictionaries.
Create linked inspection and defect records
The models describe valid data. The next operations turn that data into stored records with generated identifiers and UTC timestamps.
- In main.py, find the health_check() function.
- Add the inspection creation operation below that function:
@app.post("/inspections", response_model=InspectionRecord)
def create_inspection(payload: InspectionCreate) -> InspectionRecord:
inspection = InspectionRecord(
id=uuid4(),
inspected_at=datetime.now(timezone.utc),
**payload.model_dump(),
)
inspections[inspection.id] = inspection
return inspection
How is an inspection created?
- The operation validates the request against InspectionCreate.
- The API generates an identifier plus an aware UTC timestamp.
- The model_dump call copies the validated request fields into InspectionRecord.
- The finished record is stored under its generated identifier.
- Save main.py.
- Return to the interactive documentation at http://127.0.0.1:8000/docs.
- Refresh the documentation page.
- Expand the POST /inspections operation.
- Activate the request editor.
- Replace the vehicle_id value with VH-2048.
- Replace the station value with paint_shop.
- Replace the inspector value with Amina Patel.
- Run the inspection operation.
The response contains a generated id plus an inspected_at timestamp. It also returns paint_shop as the selected station.
- Record the returned inspection identifier here: your inspection UUID.
Good progress. Your first plant inspection now has a generated identity and a precise inspection time.
Is the inspection request rejected?
Confirm that the request uses the three field names from InspectionCreate. Check that station contains one of the enum values shown in the documentation.
Help me troubleshoot my inspection request.
- In main.py, find the final line of create_inspection().
- Add the defect creation operation below that function:
@app.post("/defects", response_model=DefectRecord)
def create_defect(payload: DefectCreate) -> DefectRecord:
if payload.inspection_id not in inspections:
raise HTTPException(status_code=404, detail="Inspection not found")
defect = DefectRecord(
id=uuid4(),
status=DefectStatus.OPEN,
audit_trail=[
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=payload.reported_by,
action="Defect reported",
to_status=DefectStatus.OPEN,
)
],
**payload.model_dump(),
)
defects[defect.id] = defect
return defect
How is the link protected?
- The operation checks that the supplied inspection identifier exists.
- A missing inspection produces a clear HTTP error.
- Every valid defect starts with open status.
- The first audit event records the reporter plus the UTC report time.
- Save main.py.
- Refresh the interactive documentation page.
- Expand the POST /defects operation.
- Activate the request editor.
- Replace the inspection_id value with your inspection UUID.
- Replace the title value with Paint blister on hood.
- Replace the description value with Raised paint blister near front edge.
The inspection link in action
The inspection identifier connects this paint defect to the paint-shop inspection you just created. The API rejects identifiers that do not exist in its inspection store.
- Replace the severity value with high.
- Replace the reported_by value with Amina Patel.
- Run the defect operation.
The response shows high severity plus open status. Its audit_trail contains a Defect reported event attributed to Amina Patel.
- Record the returned defect identifier here: your defect UUID.
That link is working. The defect now carries its inspection reference plus the first event in its history.
Is the linked defect rejected?
Confirm that inspection_id matches the identifier returned by the inspection operation. Check that severity uses the lowercase enum value.
Help me troubleshoot my linked defect request.
Retrieve the defect and verify the link
A quality engineer needs to retrieve one defect by its identifier. This operation proves that the stored record preserves its linked inspection, severity, initial status, and first audit event.
- In main.py, find the final line of create_defect().
- Add the defect retrieval operation below that function:
@app.get("/defects/{defect_id}", response_model=DefectRecord)
def read_defect(defect_id: UUID) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
return defect
How does retrieval work?
- The path supplies the defect identifier as a UUID.
- The operation looks up that identifier in the defects dictionary.
- An unknown identifier produces a clear HTTP error.
- A known identifier returns the complete DefectRecord.
- Save main.py.
- Refresh the interactive documentation page.
- Expand the GET /defects/{defect_id} operation.
- Activate the path input.
- Enter your defect UUID as the defect identifier.
Before you run the retrieval, which inspection reference and workflow status do you expect the stored defect to contain?
- Run the defect retrieval operation.
The retrieved defect contains your inspection UUID as its inspection_id. It also shows high severity plus open status.
The inspection response showed paint_shop as its station. The matching inspection_id proves that the retrieved defect belongs to that paint-shop inspection.
You now have the first end-to-end API slice working. A plant inspection can produce a linked quality defect with a timestamped reporting event.
Cannot retrieve the defect?
Confirm that you entered the defect identifier instead of the inspection identifier. Keep the development server running because restarting it clears both in-memory dictionaries.
Help me troubleshoot defect retrieval in my in-memory API.
✔️ Awesome, I've got everything!
Great work. Save main.py and keep the development server running for the next step.
ⓧ I'd like to double check the full code
Compare your complete main.py with this reference:
from datetime import datetime, timezone
from enum import Enum
from uuid import UUID, uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class ProductionStation(str, Enum):
BODY_SHOP = "body_shop"
PAINT_SHOP = "paint_shop"
FINAL_ASSEMBLY = "final_assembly"
class Severity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
class DefectStatus(str, Enum):
OPEN = "open"
CONTAINED = "contained"
CLOSED = "closed"
class InspectionCreate(BaseModel):
vehicle_id: str
station: ProductionStation
inspector: str
class InspectionRecord(InspectionCreate):
id: UUID
inspected_at: datetime
class DefectCreate(BaseModel):
inspection_id: UUID
title: str
description: str
severity: Severity
reported_by: str
class AuditEvent(BaseModel):
timestamp: datetime
actor: str
action: str
from_status: DefectStatus | None = None
to_status: DefectStatus
class DefectRecord(DefectCreate):
id: UUID
status: DefectStatus
containment_action: str | None = None
corrective_action: str | None = None
audit_trail: list[AuditEvent]
inspections: dict[UUID, InspectionRecord] = {}
defects: dict[UUID, DefectRecord] = {}
@app.get("/")
def health_check() -> dict[str, str]:
return {"status": "quality workflow API is running"}
@app.post("/inspections", response_model=InspectionRecord)
def create_inspection(payload: InspectionCreate) -> InspectionRecord:
inspection = InspectionRecord(
id=uuid4(),
inspected_at=datetime.now(timezone.utc),
**payload.model_dump(),
)
inspections[inspection.id] = inspection
return inspection
@app.post("/defects", response_model=DefectRecord)
def create_defect(payload: DefectCreate) -> DefectRecord:
if payload.inspection_id not in inspections:
raise HTTPException(status_code=404, detail="Inspection not found")
defect = DefectRecord(
id=uuid4(),
status=DefectStatus.OPEN,
audit_trail=[
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=payload.reported_by,
action="Defect reported",
to_status=DefectStatus.OPEN,
)
],
**payload.model_dump(),
)
defects[defect.id] = defect
return defect
@app.get("/defects/{defect_id}", response_model=DefectRecord)
def read_defect(defect_id: UUID) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
return defect
Your existing requirements.txt remains unchanged:
fastapi[standard]==0.143.0
pydantic==2.14.0
Your API now records inspections and the defects they produce. Next, you will add a status update operation and test whether typed data alone protects the factory workflow.
Expose the Broken Workflow
Your API now records inspections. It links each new defect to its source inspection.
This step adds a temporary status update to test whether a typed API protects the factory sequence. A Pydantic request model validates the shape of each update.
You will deliberately try to move a defect from open directly to closed. The response reveals whether valid fields are enough to enforce the quality workflow.
In this step, get ready to:
- Define the request model for defect status updates.
- Expose status changes through a temporary update operation.
- Test a direct close to reveal how the operation behaves.
Define the workflow update request
A status update needs the requested state plus the person responsible for the change. It also carries the two action fields that later workflow stages use.
- In main.py, locate the DefectRecord class.
- Insert the following model after DefectRecord. Place it before the inspections dictionary:
class WorkflowUpdate(BaseModel):
new_status: DefectStatus
actor: str
containment_action: str | None = None
corrective_action: str | None = None
What does this model do?
- The new_status field accepts one of the states defined by DefectStatus.
- The actor field identifies the person submitting the update.
- The optional action fields can hold evidence for containment or corrective work.
- Save main.py.
- Return to the terminal running the development server.
You should see FastAPI reload the application without a Python traceback. The new request model is now valid application code.
Seeing a reload failure?
Check that WorkflowUpdate sits below the complete DefectRecord class. Confirm that each field inside the new class uses four spaces of indentation.
Ask for help with the model error by selecting help me debug my WorkflowUpdate model.
Add the temporary status operation
The update operation receives a defect identifier in its path. It retrieves that record before copying the submitted values onto it.
- In main.py, scroll to the read_defect() function.
- Insert the following operation below the return defect line:
@app.patch("/defects/{defect_id}/status", response_model=DefectRecord)
def update_defect_status(
defect_id: UUID,
update: WorkflowUpdate,
) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
defect.status = update.new_status
defect.containment_action = update.containment_action
defect.corrective_action = update.corrective_action
return defect
What does this operation do?
- The route exposes PATCH /defects/{defect_id}/status for partial updates to one stored defect.
- The dictionary lookup finds the record associated with the supplied defect_id.
- The missing-record check returns the existing Defect not found error when the identifier has no match.
- The final assignments copy the submitted status fields onto the record before returning it.
- Save main.py.
- Refresh the interactive documentation at http://127.0.0.1:8000/docs.
You should see PATCH /defects/{defect_id}/status listed with the existing inspection and defect operations. The temporary update operation is ready for an experiment.
Missing the status operation?
Check the terminal for a syntax or indentation problem after the reload. Confirm that the decorator begins at the far left of main.py.
Ask for help by selecting help me find why my PATCH operation is missing.
✔️ Awesome, I've got everything!
Your WorkflowUpdate model and temporary status operation are saved. Keep the development server running for the direct-close experiment.
ⓧ I'd like to double check the full code
Compare your complete main.py with this version before testing the workflow.
from datetime import datetime, timezone
from enum import Enum
from uuid import UUID, uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class ProductionStation(str, Enum):
BODY_SHOP = "body_shop"
PAINT_SHOP = "paint_shop"
FINAL_ASSEMBLY = "final_assembly"
class Severity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
class DefectStatus(str, Enum):
OPEN = "open"
CONTAINED = "contained"
CLOSED = "closed"
class InspectionCreate(BaseModel):
vehicle_id: str
station: ProductionStation
inspector: str
class InspectionRecord(InspectionCreate):
id: UUID
inspected_at: datetime
class DefectCreate(BaseModel):
inspection_id: UUID
title: str
description: str
severity: Severity
reported_by: str
class AuditEvent(BaseModel):
timestamp: datetime
actor: str
action: str
from_status: DefectStatus | None = None
to_status: DefectStatus
class DefectRecord(DefectCreate):
id: UUID
status: DefectStatus
containment_action: str | None = None
corrective_action: str | None = None
audit_trail: list[AuditEvent]
class WorkflowUpdate(BaseModel):
new_status: DefectStatus
actor: str
containment_action: str | None = None
corrective_action: str | None = None
inspections: dict[UUID, InspectionRecord] = {}
defects: dict[UUID, DefectRecord] = {}
@app.get("/")
def health_check() -> dict[str, str]:
return {"status": "quality workflow API is running"}
@app.post("/inspections", response_model=InspectionRecord)
def create_inspection(payload: InspectionCreate) -> InspectionRecord:
inspection = InspectionRecord(
id=uuid4(),
inspected_at=datetime.now(timezone.utc),
**payload.model_dump(),
)
inspections[inspection.id] = inspection
return inspection
@app.post("/defects", response_model=DefectRecord)
def create_defect(payload: DefectCreate) -> DefectRecord:
if payload.inspection_id not in inspections:
raise HTTPException(status_code=404, detail="Inspection not found")
defect = DefectRecord(
id=uuid4(),
status=DefectStatus.OPEN,
audit_trail=[
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=payload.reported_by,
action="Defect reported",
to_status=DefectStatus.OPEN,
)
],
**payload.model_dump(),
)
defects[defect.id] = defect
return defect
@app.get("/defects/{defect_id}", response_model=DefectRecord)
def read_defect(defect_id: UUID) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
return defect
@app.patch("/defects/{defect_id}/status", response_model=DefectRecord)
def update_defect_status(
defect_id: UUID,
update: WorkflowUpdate,
) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
defect.status = update.new_status
defect.containment_action = update.containment_action
defect.corrective_action = update.corrective_action
return defect
What should you compare?
The new WorkflowUpdate model belongs between DefectRecord and the in-memory dictionaries. The temporary update operation belongs after read_defect().
Your requirements.txt remains unchanged:
fastapi[standard]==0.143.0
pydantic==2.14.0
Why are the requirements unchanged?
This step uses the installed FastAPI and Pydantic packages. No additional dependency is needed for the temporary status operation.
Run the direct-close experiment
Reloading main.py restarted the in-memory application state. Create a fresh inspection plus a linked defect before testing the update.
- Return to the interactive documentation at http://127.0.0.1:8000/docs.
- Expand POST /inspections.
- Enable the request editor using the control in the operation panel.
- Set vehicle_id to VIN-BROKEN-001.
- Set station to paint_shop.
- Set inspector to Maya Chen.
- Submit the inspection request using the control beneath the request body.
- Record the returned id as your inspection UUID.
The inspection response gives you a fresh UUID. That value links the next defect to its paint-shop inspection.
- Expand POST /defects.
- Enable the request editor using the control in the operation panel.
- Replace the inspection_id value with your inspection UUID.
- Set title to Paint blister.
- Set description to Blistering near the rear door.
- Set severity to high.
- Set reported_by to Maya Chen.
- Submit the defect request using the control beneath the request body.
The response shows status set to open. Its audit trail starts with the Defect reported event.
Your clean test record
This new defect begins at the correct starting state. Its empty action fields make the direct-close test easy to inspect.
- Record the returned id as your defect UUID.
- Expand PATCH /defects/{defect_id}/status.
- Enable the request editor using the control in the operation panel.
- Enter your defect UUID in the defect_id path field.
- Set new_status to closed.
- Set actor to Maya Chen.
- Leave containment_action as null.
- Leave corrective_action as null.
Before you submit the request, what do you think the API does with this direct jump? Make your prediction now.
- Submit the status request using the control beneath the request body.
You'll see a successful response with status set to closed. Both action fields remain null.
The audit trail still has only Defect reported leading to open. You have exposed the intended shortfall: valid types accepted an invalid quality process.
That experiment exposed the missing guard. Next, you'll enforce the ordered path from open through containment to closure.
Enforce the Workflow and Audit Trail
Your temporary FastAPI endpoint exposed the gap between typed data and a protected quality process. It accepted a direct jump from an open defect to a closed defect.
The fix turns the factory sequence into business-rule validation. An audit trail then records the evidence behind every accepted change.
In this step, get ready to:
- Permit only the next valid status in the defect workflow.
- Require containment evidence before closure evidence.
- Return HTTP errors for invalid moves while preserving accepted changes in the audit trail.
Encode the allowed transitions
A transition map gives each current status one permitted destination. Any requested status outside that path becomes a rejected workflow move.
- In main.py, return to update_defect_status().
- Find defect.status = update.new_status.
- Extend the selection through defect.corrective_action = update.corrective_action.
- Replace the three selected lines with this transition validation:
allowed_transitions: dict[DefectStatus, DefectStatus] = {
DefectStatus.OPEN: DefectStatus.CONTAINED,
DefectStatus.CONTAINED: DefectStatus.CLOSED,
}
expected_status = allowed_transitions.get(defect.status)
if expected_status is None:
raise HTTPException(status_code=400, detail="Closed defects cannot be changed")
if update.new_status != expected_status:
raise HTTPException(
status_code=400,
detail=(
f"Status must move from {defect.status.value} "
f"to {expected_status.value}"
),
)
What does this code do?
- The allowed_transitions map permits open to move only to contained.
- The same map permits contained to move only to closed.
- A closed defect has no next status, so the endpoint rejects further changes.
- A request for any unexpected status returns an HTTPException with status code 400.
- Save main.py.
- Return to http://127.0.0.1:8000/docs from earlier.
- Expand POST /inspections.
- Submit the fictional paint-shop inspection request from Step 2 again.
- Store the returned inspection ID here: your rule-test inspection UUID.
Saving the file reloads the development server. That reload clears the in-memory records, so this inspection gives your updated endpoint fresh data.
- Expand POST /defects.
- Paste your rule-test inspection UUID into inspection_id.
- Submit the high-severity paint defect request from Step 2 again.
- Store the returned defect ID here: your rule-test defect UUID.
Before you submit the next update, do you think the API still accepts a direct move from open to closed?
- Expand PATCH /defects/{defect_id}/status.
- Paste your rule-test defect UUID into defect_id.
- Set new_status to closed.
- Set actor to your quality engineer name.
- Remove the containment_action field from the request body.
- Remove the corrective_action field from the request body.
- Submit the status update.
You receive an HTTP 400 response with Status must move from open to contained. That rejection proves the direct-close gap is sealed.
Does the direct close still succeed?
Confirm that you replaced all three temporary assignment lines. Make sure the original return defect remains below the new validation code.
Save main.py again if the development server has not reloaded your latest changes.
Help me debug the transition validation.
Require action evidence and record changes
A valid destination still needs evidence. Containment records the immediate control while corrective action records the lasting fix.
- In update_defect_status(), find the final return defect line.
- Replace that line with the action checks and audit update below:
if update.new_status == DefectStatus.CONTAINED:
if not update.containment_action:
raise HTTPException(
status_code=400,
detail="A containment action is required",
)
defect.containment_action = update.containment_action
action = "Defect contained"
else:
if not update.corrective_action:
raise HTTPException(
status_code=400,
detail="A corrective action is required",
)
defect.corrective_action = update.corrective_action
action = "Corrective action recorded and defect closed"
previous_status = defect.status
defect.status = update.new_status
defect.audit_trail.append(
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=update.actor,
action=action,
from_status=previous_status,
to_status=update.new_status,
)
)
return defect
How does the evidence become history?
- Entering contained requires a containment_action before the defect changes.
- Entering closed requires a corrective_action before the defect changes.
- The endpoint captures previous_status before assigning the new status.
- Each accepted update appends an AuditEvent with an aware UTC timestamp.
- The event preserves the actor, action text, previous status, and new status.
- Save main.py.
- Return to the terminal running fastapi dev.
- Confirm the server reload completes without a Python traceback.
- Refresh the documentation page.
You should still see PATCH /defects/{defect_id}/status in the interactive documentation. The endpoint is ready for a complete workflow test.
Seeing a Python error after saving?
Check that the new block stays inside update_defect_status() with four spaces of indentation.
Confirm that both closing parentheses remain below the AuditEvent fields.
Help me fix the workflow update code.
Run the protected workflow
The latest reload cleared the earlier test records. A fresh defect now lets you test rejection followed by the two accepted transitions on one record.
- Expand POST /inspections in the documentation.
- Submit the same fictional paint-shop inspection request again.
- Store the returned ID here: your final inspection UUID.
The returned UUID identifies the inspection that anchors your final workflow demonstration.
- Expand POST /defects.
- Paste your final inspection UUID into inspection_id.
- Submit the same high-severity paint defect request again.
- Store the returned defect ID here: your final defect UUID.
Before you test the protected route, predict whether this new open defect can skip containment.
- Expand PATCH /defects/{defect_id}/status.
- Paste your final defect UUID into defect_id.
- Set new_status to closed.
- Set actor to your quality engineer name.
- Remove both action fields from the request body.
- Submit the direct-close request.
You receive HTTP 400 again. The defect remains open because rejected updates never reach the status assignment.
- Change new_status to contained.
- Keep actor set to your quality engineer name.
- Add containment_action with this value: describe how the vehicle was contained.
- Submit the containment update.
You should see status set to contained. The audit trail now contains Defect reported followed by Defect contained.
Before you submit the closure, predict how many audit events the final response should contain.
- Change new_status to closed.
- Keep actor set to your quality engineer name.
- Remove containment_action from the request body.
- Add corrective_action with this value: describe the lasting corrective action.
- Submit the closure update.
You should see status set to closed with both action fields saved. The three ordered audit events are Defect reported, Defect contained, and Corrective action recorded and defect closed.
That is the full quality safeguard working. Invalid jumps stop at the boundary while accepted changes leave a timestamped history.
✔️ Awesome, I've got everything!
- Save main.py before finishing the step.
- Keep the final closed-defect response available for your checkpoint.
ⓧ I'd like to double check the full code
Your complete main.py should match this file.
from datetime import datetime, timezone
from enum import Enum
from uuid import UUID, uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class ProductionStation(str, Enum):
BODY_SHOP = "body_shop"
PAINT_SHOP = "paint_shop"
FINAL_ASSEMBLY = "final_assembly"
class Severity(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
class DefectStatus(str, Enum):
OPEN = "open"
CONTAINED = "contained"
CLOSED = "closed"
class InspectionCreate(BaseModel):
vehicle_id: str
station: ProductionStation
inspector: str
class InspectionRecord(InspectionCreate):
id: UUID
inspected_at: datetime
class DefectCreate(BaseModel):
inspection_id: UUID
title: str
description: str
severity: Severity
reported_by: str
class AuditEvent(BaseModel):
timestamp: datetime
actor: str
action: str
from_status: DefectStatus | None = None
to_status: DefectStatus
class DefectRecord(DefectCreate):
id: UUID
status: DefectStatus
containment_action: str | None = None
corrective_action: str | None = None
audit_trail: list[AuditEvent]
class WorkflowUpdate(BaseModel):
new_status: DefectStatus
actor: str
containment_action: str | None = None
corrective_action: str | None = None
inspections: dict[UUID, InspectionRecord] = {}
defects: dict[UUID, DefectRecord] = {}
@app.get("/")
def health_check() -> dict[str, str]:
return {"status": "quality workflow API is running"}
@app.post("/inspections", response_model=InspectionRecord)
def create_inspection(payload: InspectionCreate) -> InspectionRecord:
inspection = InspectionRecord(
id=uuid4(),
inspected_at=datetime.now(timezone.utc),
**payload.model_dump(),
)
inspections[inspection.id] = inspection
return inspection
@app.post("/defects", response_model=DefectRecord)
def create_defect(payload: DefectCreate) -> DefectRecord:
if payload.inspection_id not in inspections:
raise HTTPException(status_code=404, detail="Inspection not found")
defect = DefectRecord(
id=uuid4(),
status=DefectStatus.OPEN,
audit_trail=[
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=payload.reported_by,
action="Defect reported",
to_status=DefectStatus.OPEN,
)
],
**payload.model_dump(),
)
defects[defect.id] = defect
return defect
@app.get("/defects/{defect_id}", response_model=DefectRecord)
def read_defect(defect_id: UUID) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
return defect
@app.patch("/defects/{defect_id}/status", response_model=DefectRecord)
def update_defect_status(
defect_id: UUID,
update: WorkflowUpdate,
) -> DefectRecord:
defect = defects.get(defect_id)
if defect is None:
raise HTTPException(status_code=404, detail="Defect not found")
allowed_transitions: dict[DefectStatus, DefectStatus] = {
DefectStatus.OPEN: DefectStatus.CONTAINED,
DefectStatus.CONTAINED: DefectStatus.CLOSED,
}
expected_status = allowed_transitions.get(defect.status)
if expected_status is None:
raise HTTPException(status_code=400, detail="Closed defects cannot be changed")
if update.new_status != expected_status:
raise HTTPException(
status_code=400,
detail=(
f"Status must move from {defect.status.value} "
f"to {expected_status.value}"
),
)
if update.new_status == DefectStatus.CONTAINED:
if not update.containment_action:
raise HTTPException(
status_code=400,
detail="A containment action is required",
)
defect.containment_action = update.containment_action
action = "Defect contained"
else:
if not update.corrective_action:
raise HTTPException(
status_code=400,
detail="A corrective action is required",
)
defect.corrective_action = update.corrective_action
action = "Corrective action recorded and defect closed"
previous_status = defect.status
defect.status = update.new_status
defect.audit_trail.append(
AuditEvent(
timestamp=datetime.now(timezone.utc),
actor=update.actor,
action=action,
from_status=previous_status,
to_status=update.new_status,
)
)
return defect
This file now contains the transition map, action requirements, status mutation, and audit append logic used in your final test.
Your unchanged requirements.txt should match these pinned dependencies.
fastapi[standard]==0.143.0
pydantic==2.14.0
These exact pins preserve the project environment you set up in Step 1.
Secret mission
Filter the Defect Queue
Turn the stored defects into a focused work queue. Add optional filters for production station, severity, and workflow status so quality engineers can find the records that need attention.
Clean Up Your Resources
Clean Up Your Resources
All of your work runs on your Linux computer. This project has no ongoing costs.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The FastAPI development server running at http://127.0.0.1:8000 with interactive documentation at http://127.0.0.1:8000/docs.
- The quality-defect-workflow project folder on your Desktop. This folder includes main.py plus requirements.txt.
- The .venv virtual environment inside the project folder. It contains FastAPI 0.143.0 plus Pydantic 2.14.0.
Keep everything running
No action needed. Choose this if you are still testing the workflow or experimenting with defect filters.
- Leave the FastAPI development server running at http://127.0.0.1:8000.
- Keep the quality-defect-workflow folder on your Desktop.
- Continue using http://127.0.0.1:8000/docs to create inspections or test the defect queue.
Pause - I'll come back to this later
Pausing stops the local server to free your terminal while keeping every project file. The API holds its inspection records plus defect records in memory, so pausing clears the current records.
- Switch back to the terminal where the FastAPI development server is running.
- Press Ctrl+C to stop the server.
The documentation URL stops responding after the server closes. Your code plus the virtual environment remain inside quality-defect-workflow.
- Reactivate .venv with the activation command from Step 1 when you return.
- Restart the FastAPI development server with the development command from Step 1.
Delete - I don't want to use this again
Deleting the project folder permanently removes main.py, requirements.txt, plus .venv. No cloud resources need separate cleanup.
- Switch back to the terminal where the FastAPI development server is running.
- Press Ctrl+C to stop the server.
- Delete the quality-defect-workflow folder from your Desktop by running these commands:
cd ~/Desktop
rm -rf quality-defect-workflow
ls
What does this cleanup do?
- The first command moves the terminal to your Desktop.
- The second command permanently removes the quality-defect-workflow folder.
- The final command lists the remaining Desktop contents. You should no longer see quality-defect-workflow.
Nice Work!
Nice Work!
Fantastic work! Your local FastAPI backend now turns a vehicle-quality process into a controlled workflow with traceable history.
You've learned how to:
- Build linked inspection and defect records around a generated inspection identifier. Preserve station context plus severity on every reported defect.
- Reveal the risk in a generic status update by closing an open defect directly. Enforce ordered workflow transitions that require containment evidence before closure.
- Return clear HTTP errors when a record is missing or a transition is invalid. Preserve every accepted change in an actor-attributed UTC audit trail.
- Secret Mission: Extend the API with a filterable defect queue. Filter by station. Filter by severity. Filter by workflow status. Combine supplied filters so every result matches.
Ready to quiz yourself?