Build a Customer Data Integration API

Introduction

30 Second Summary

Every company has messy data. Someone has to take that chaos and turn it into something the rest of the team can actually use.

In this project, you will build a data integration API that ingests messy customer product data, validates it against strict rules, and serves clean results. You will follow the Forward Deployment Engineering workflow: scope a customer problem, build the integration, hit a data quality failure, and fix it.

What You'll Build

Picture opening your browser, throwing a batch of inconsistent product records at your API, and instantly getting back a clean separation: valid products stored and queryable, broken records returned with plain-English explanations of exactly what went wrong.

By the end of this project, you'll have:

• A running REST API with auto-generated interactive docs where you can test every endpoint live in your browser.
• A partial-success ingestion system that saves valid records and returns clear error reports for bad ones. No more crashing on the first broken row.
• Filtered product queries that let you search by category and price range, proving your integrated data is clean and usable.
• Secret Mission: A data quality dashboard endpoint that turns ingestion failures into actionable recommendations for the customer.

Do I need any experience before starting?

Basic Python knowledge (variables, functions, lists, dictionaries) and basic terminal usage is all you need. No prior API or web development experience required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Project and Install FastAPI

To build NovaMart's data integration API, you need a running server that can receive requests and send back responses. FastAPI is the go-to framework for production Python APIs and appears in the majority of FDE job postings.

Your project folder is ready and FastAPI is installed. Now it's time to write your first endpoint and confirm everything is wired up correctly.

In this step, get ready to:

• Create a minimal FastAPI app with a root endpoint.
• Start the development server and verify it runs in your browser.

Open your project in the IDE

• Switch to Antigravity IDE (or VS Code).
• Open the project folder you created earlier by clicking File in the top menu bar.
• Click Open Folder.
• Select the project folder you created on your Desktop, then click Open.

Create your first FastAPI endpoint

Every FastAPI app starts with a Python file that creates an app instance and defines at least one endpoint. An endpoint is a URL path that your server responds to when a client sends a request.

• Create a new file called main.py in your project folder and add this code:

from fastapi import FastAPI

# Create the FastAPI app with metadata for the docs page
app = FastAPI(
    title="NovaMart Product Integration API",
    description="Data integration API for NovaMart's product catalog",
    version="1.0.0",
)

# Define a root endpoint that returns a welcome message
@app.get("/")
def read_root():
    return {"message": "NovaMart Product Integration API", "docs": "/docs"}

What does this code do?

• from fastapi import FastAPI imports the framework so you can create your app.
• app = FastAPI(...) creates your API application with a title, description, and version. These appear on the auto-generated documentation page.
• @app.get("/") registers a route that responds to GET requests at the root URL. When someone visits your server, this function runs and returns a JSON message.

• Save main.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py file. Compare it with yours character by character.

from fastapi import FastAPI

app = FastAPI(
    title="NovaMart Product Integration API",
    description="Data integration API for NovaMart's product catalog",
    version="1.0.0",
)

@app.get("/")
def read_root():
    return {"message": "NovaMart Product Integration API", "docs": "/docs"}

Start the development server

Before you run this, what do you think will happen when you start the server? Will it print anything to the terminal?

• In your terminal (with your virtual environment still active), start the FastAPI development server by running this command:

fastapi dev main.py

Your terminal prints startup information including the server URL and confirms the application started successfully. You should see a line showing http://127.0.0.1:8000 and a message that says Application startup complete.

What does this command do?

The fastapi dev command detects your FastAPI app in main.py, starts a Uvicorn web server, and enables auto-reload.

Auto-reload means the server restarts itself every time you save a code change. You will not need to stop and restart manually during this project.

• Open your web browser and go to http://127.0.0.1:8000/docs.

You should see the interactive Swagger UI page with your app title (NovaMart Product Integration API) at the top and a single GET / endpoint listed. This documentation page is auto-generated by FastAPI from your code.

Don't see the Swagger UI page?

• Check that your terminal still shows the server running. If you see an error, make sure your virtual environment is active (you should see (venv) at the start of your prompt).
• Confirm the URL is exactly http://127.0.0.1:8000/docs with no typos.
• If you see ModuleNotFoundError, FastAPI may not be installed in your active environment. Run pip install "fastapi[standard]" and try again.

Why isn't my FastAPI server starting?

Your API is alive and serving documentation. Next up, you'll define data models and build real endpoints that accept and return product data.

Build the Basic Product API with Pydantic Validation

In Step 1, you got your project running and confirmed the dev server works. Now it's time to define what "clean data" actually looks like for NovaMart's product catalog.

Before you can handle messy data, you need endpoints that accept products and serve them back. This step gives you an immediate visible win: a running API you can interact with directly in the browser.

In this step, get ready to:

• Define Pydantic models that enforce strict data rules on product submissions.
• Create POST and GET endpoints to ingest and retrieve products.
• Test your API live through the interactive Swagger UI docs.

Define the data models

Every product NovaMart sends needs to meet strict rules before your API accepts it. Pydantic models let you declare those rules as Python type annotations, and FastAPI automatically rejects any request that breaks them.

• In main.py, find the import line at the top of your file:

from fastapi import FastAPI

• Replace that line with these three imports:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator
from datetime import datetime

What do these imports do?

• HTTPException lets you return specific HTTP error codes (like 409 for duplicates).
• BaseModel, Field, and field_validator from Pydantic define your data rules.
• datetime timestamps when each product gets ingested.

• Below the app = FastAPI(...) block and above @app.get("/"), add the ProductInput model by pasting the following code:

# --- Data Models ---

class ProductInput(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    price: float = Field(gt=0)
    category: str = Field(min_length=1)
    sku: str = Field(pattern=r"^[A-Z]{2,5}-\d{3,6}$")

    @field_validator("name")
    @classmethod
    def name_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be blank")
        return v.strip()

What does this model enforce?

• name must be 1-200 characters and cannot be only whitespace.
• price must be greater than zero. No free or negative-priced products.
• category must have at least one character.
• sku must match a specific pattern: 2-5 uppercase letters, a dash, then 3-6 digits (e.g. ELEC-001).
• The name_not_empty validator adds an extra check that catches names made entirely of spaces.

• Save main.py.
• Check the terminal where your dev server is running. You should see a reload message confirming the server picked up your changes without errors.

Seeing an import error or server crash?

• Make sure your virtual environment is still active. You should see (venv) in your terminal prompt.
• Check that field_validator is spelled exactly as shown. A common mistake is writing validator instead.
• Verify the regex pattern string uses a raw prefix: r"^[A-Z]{2,5}-\d{3,6}$". Missing the r causes escape character issues.

Help me fix an import error in my FastAPI app after adding Pydantic models

• Directly below the ProductInput class, add the ProductOutput model and in-memory storage by pasting this code:

class ProductOutput(BaseModel):
    name: str
    price: float
    category: str
    sku: str
    ingested_at: str


# --- In-Memory Storage ---

products_db: dict[str, ProductOutput] = {}
last_ingestion_time: str | None = None


# --- Endpoints ---

What does this code do?

ProductOutput defines the shape of data your API sends back. It includes an ingested_at timestamp so you know when each product was added.

products_db is a dictionary that stores products in memory, keyed by SKU. The # --- Endpoints --- comment goes directly above your existing @app.get("/") function to keep the file organized.

All data disappears when you stop the server. No database setup needed.

• Save main.py. Check the terminal to confirm the server reloads successfully.

Create the product endpoints

With your models defined, you can now build the endpoints that use them. The POST endpoint ingests a product, and the GET endpoint retrieves all stored products.

• In main.py, below the read_root function, add the POST endpoint by pasting this code:

@app.post("/products", response_model=ProductOutput, status_code=201)
def create_product(product: ProductInput):
    global last_ingestion_time
    # Reject duplicate SKUs
    if product.sku in products_db:
        raise HTTPException(status_code=409, detail=f"Product with SKU {product.sku} already exists")
    now = datetime.now().isoformat()
    stored = ProductOutput(
        name=product.name,
        price=product.price,
        category=product.category,
        sku=product.sku,
        ingested_at=now,
    )
    products_db[product.sku] = stored
    last_ingestion_time = now
    return stored

What does this endpoint do?

• It accepts a JSON body matching ProductInput. FastAPI validates the body automatically before your function runs.
• If the SKU already exists, it raises a 409 Conflict error instead of silently overwriting.
• It creates a ProductOutput with a timestamp, stores it in products_db, and returns the clean result with a 201 Created status.

• Save main.py.
• Switch back to your browser at http://127.0.0.1:8000/docs and refresh the page.
• You should see a new green POST /products endpoint listed alongside your original root endpoint.

Don't see the new endpoint?

• Hard-refresh your browser with Cmd+Shift+R (macOS) or Ctrl+Shift+R (Windows).
• Check the terminal. If it shows a syntax error, look for missing colons, unmatched parentheses, or incorrect indentation in the code you just added.
• Make sure the @app.post decorator is at the same indentation level as @app.get("/") above it.

Why isn't my new POST endpoint showing up in the Swagger docs?

• Below the create_product function, add the GET endpoint by pasting this code:

@app.get("/products", response_model=list[ProductOutput])
def list_products():
    return list(products_db.values())

What does this endpoint do?

It returns every product currently stored in products_db as a JSON list. The response_model=list[ProductOutput] tells FastAPI to document the response shape in the Swagger UI.

• Save main.py and refresh your browser.
• You should now see both POST /products and GET /products listed in the Swagger UI.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and both endpoints appear in your Swagger UI.

ⓧ I'd like to double check the full code

Your main.py should now contain these sections in order. Scroll up and confirm each one is present:

• Three import lines (fastapi, pydantic, datetime).
• The app = FastAPI(...) block.
• # --- Data Models --- section with ProductInput (including the name_not_empty validator) and ProductOutput.
• # --- In-Memory Storage --- section with products_db and last_ingestion_time.
• # --- Endpoints --- section with read_root, create_product, and list_products.

Test your API through the interactive docs

Your endpoints are live. Time to send real data through them and see validation in action.

• In the Swagger UI at http://127.0.0.1:8000/docs, click on the green POST /products bar to expand it.
• Click the Try it out button on the right side.
• Replace the example JSON in the request body with this valid product:

{
    "name": "Wireless Headphones",
    "price": 79.99,
    "category": "electronics",
    "sku": "ELEC-001"
}

• Click Execute.

You should see a 201 response with your product returned as clean JSON, including an ingested_at timestamp.

What just happened?

FastAPI validated your JSON against the ProductInput model automatically. The name, price, category, and SKU all passed the rules you defined. The product is now stored in memory.

Now test the GET endpoint. Before you try it, what do you expect to see when you retrieve all products?

• Click on the blue GET /products bar to expand it.
• Click Try it out, then click Execute.

You should see a 200 response containing a JSON list with the Wireless Headphones product you just created. The data comes back clean and typed, exactly matching your ProductOutput model.

Your data rules are already working

Try POSTing a product with a negative price or an empty name. FastAPI will reject it automatically with a 422 error. You defined "clean" with your Pydantic model, and FastAPI enforces it on every request.

You have a working API that validates incoming data and serves it back clean. Next up, you will throw messy real-world data at it and watch it break.

Ingest NovaMart's Messy Data and Watch It Break

Your single-product endpoint works perfectly when you send clean, well-formatted data. That is exactly the environment a demo lives in.

Production is different. NovaMart's real product catalog has blank names, negative prices, and SKU formats that follow no standard. When this data hits your API, you will see firsthand why Forward Deployment Engineering work is hard: the gap between "works in a demo" and "survives real customer data."

In this step, get ready to:

• Create a simulated NovaMart data file with realistic messy records.
• Add a batch ingestion endpoint that accepts multiple products at once.
• POST the messy data and observe how your API handles it.

Create the NovaMart data file

Every FDE integration starts with sample data from the customer. NovaMart has sent you a JSON export of their product catalog. Some records are clean. Others have problems that are typical of real-world data exports.

• Create a new file called novamart_data.json in your novamart-api project folder (the same folder where main.py lives).
• Paste the following product records into the file:

[
    {"name": "Wireless Headphones", "price": 79.99, "category": "electronics", "sku": "ELEC-001"},
    {"name": "Running Shoes", "price": 129.50, "category": "footwear", "sku": "FOOT-042"},
    {"name": "", "price": 19.99, "category": "accessories", "sku": "ACC-100"},
    {"name": "Yoga Mat", "price": -15.00, "category": "fitness", "sku": "FIT-007"},
    {"name": "Coffee Mug", "price": 12.99, "category": "kitchen", "sku": "bad-sku"},
    {"name": "Desk Lamp", "price": 45.00, "category": "office", "sku": "OFF-200"},
    {"name": "Phone Case", "price": 0, "category": "electronics", "sku": "ELEC-002"},
    {"name": "Water Bottle", "price": 24.99, "category": "fitness", "sku": "FIT-008"}
]

What is wrong with this data?

Four of these eight records are intentionally broken. This mirrors what real customer exports look like.

• ACC-100: empty name (blank string).
• FIT-007: negative price (-15.00).
• bad-sku (Coffee Mug): SKU does not match the required format ^[A-Z]{2,5}-\d{3,6}$.
• ELEC-002 (Phone Case): zero price (must be greater than 0).

• Save novamart_data.json.
• Confirm the file exists by running:

ls

You should see novamart_data.json listed alongside main.py.

Don't see the file?

• Make sure you created the file inside your novamart-api folder, not in a subfolder or your home directory.
• Check the filename spelling. It must be exactly novamart_data.json with an underscore, not a hyphen.

I created novamart_data.json but it's not showing up when I run ls in my project folder.

Add a batch ingestion endpoint

NovaMart does not send products one at a time. They export entire catalogs in bulk. You need an endpoint that accepts a list of products in a single request.

• In main.py, find the create_product function's closing return stored line.
• Add the following batch endpoint below it (before the list_products function):

@app.post("/products/batch")
def create_products_batch(products: list[dict]):
    # Naive implementation - will fail on bad data
    results = []
    for record in products:
        product = ProductInput(**record)
        now = datetime.now().isoformat()
        stored = ProductOutput(
            name=product.name,
            price=product.price,
            category=product.category,
            sku=product.sku,
            ingested_at=now,
        )
        products_db[product.sku] = stored
        results.append(stored)
    return results

What does this code do?

• products: list[dict] accepts a JSON array of objects with no upfront type checking from FastAPI.
• The loop manually validates each record by passing it to ProductInput(**record). If any record is invalid, Pydantic raises an exception.
• There is no error handling. If one bad record appears, the entire request crashes mid-loop.

• Save main.py.
• Switch back to your browser at http://127.0.0.1:8000/docs and refresh the page.

You should see a new POST /products/batch endpoint listed in the Swagger UI alongside your existing endpoints.

Don't see the new endpoint?

• Make sure you saved main.py. The dev server auto-reloads on save.
• Check your terminal for error messages. A syntax error will prevent the server from reloading.
• Confirm the new function is placed between create_product and list_products, not indented inside another function.

Why isn't my new endpoint showing up in the Swagger docs?

POST the messy batch and watch it break

This is the moment of truth. You have an endpoint that accepts bulk data and a file full of messy records. Time to see what happens when production-quality data meets a naive implementation.

• In the Swagger UI, click on the POST /products/batch row to expand it.
• Click Try it out.
• Delete the placeholder content in the Request body text area.
• Open novamart_data.json in your editor, select all the contents, and copy them.
• Paste the JSON array into the Request body field in the Swagger UI.

Before you hit Execute, think about this: your batch has 8 records and 4 of them are invalid. Do you think the request will succeed, partially succeed, or fail entirely?

• Click Execute.

The server returns a 500 Internal Server Error. The response body in Swagger UI shows a generic error message with no useful detail about which record failed or why.

Why does this happen?

The loop processes records one by one. The first two records (ELEC-001 and FOOT-042) are valid, so they pass. When the loop reaches the third record (ACC-100 with a blank name), ProductInput(**record) raises a Pydantic ValidationError. Because there is no try/except block, the exception crashes the entire request.

FastAPI only returns helpful 422 validation errors when IT handles the validation (through typed request parameters). When you raise a ValidationError manually inside your function body, FastAPI treats it as an unhandled server error and returns a generic 500.

• Check your terminal where the dev server is running.

You will see a full Python traceback with Pydantic validation details. The error names the failing field but buries it in technical jargon that no NovaMart product manager would understand.

This failure is intentional

You just experienced the core FDE problem. Your API works in a sandbox with clean data. The moment real customer data arrives, it crashes without grace.

One bad record kills the entire batch. The four valid records (Wireless Headphones, Running Shoes, Desk Lamp, Water Bottle) are lost to the client along with the bad ones. The client has no idea what went wrong or how to fix it. The next step fixes this with production-grade error handling.

✔️ Awesome, I've got everything!

Great. Make sure you have saved both novamart_data.json and main.py.

ⓧ I'd like to double check the full code

Your main.py should contain these sections in order:

• Imports (FastAPI, HTTPException, BaseModel, Field, field_validator, datetime).
• App creation with title, description, and version.
• ProductInput model with four fields and the name_not_empty validator.
• ProductOutput model.
• In-memory storage: products_db and last_ingestion_time.
• GET / root endpoint.
• POST /products (single product) endpoint.
• POST /products/batch endpoint with the naive loop (no error handling).
• GET /products endpoint.

Your novamart_data.json should contain a JSON array with exactly 8 product objects.

Your API crashes at the first sign of bad data, taking perfectly valid records down with it. In the next step, you will build production-grade error handling that saves the good records and reports exactly what is wrong with the bad ones.

Build Production-Grade Error Handling

In the previous step, you saw the core FDE problem firsthand. One bad record in NovaMart's batch killed the entire request, and the valid products were lost along with the invalid ones.

Now you will build a partial-success pattern. Good records get accepted and stored. Bad records get rejected with clear, human-readable error messages. This is what separates a working demo from a production deployment.

In this step, get ready to:

• Define response models that separate accepted products from rejected records.
• Rewrite the batch endpoint to validate each record individually with try/except.
• Test the resilient endpoint with NovaMart's messy data and see structured error reports.

Define the response models

Before you rewrite the endpoint logic, you need data models that describe the shape of a partial-success response. One model holds rejected records with their error reasons. The other wraps the full response with accepted products, rejected records, and a count summary.

• In main.py, find the ProductOutput class. Add the following two new models directly below it:

class RejectedRecord(BaseModel):
    # Holds the raw data that failed and why
    record: dict
    reasons: list[str]


class BatchResponse(BaseModel):
    # Structured response for partial-success ingestion
    accepted: list[ProductOutput]
    rejected: list[RejectedRecord]
    summary: dict

What do these models do?

RejectedRecord pairs each failed record with a list of human-readable reasons explaining what went wrong. The original record is preserved as a raw dict so the customer can see exactly which data was rejected.

BatchResponse gives the caller a complete picture: which products made it in, which were rejected, and a quick summary count. No data is silently dropped.

• Save main.py.

Rewrite the batch endpoint to validate each record individually

The naive endpoint tried to validate the entire list at once. Your new version wraps each record in a try/except so that one bad record never kills the rest of the batch.

• In main.py, find the existing create_products_batch function. Delete the entire function (from @app.post("/products/batch") to the end of its body) and replace it with the following code:

@app.post("/products/batch", response_model=BatchResponse)
def create_products_batch(products: list[dict]):
    global last_ingestion_time
    accepted: list[ProductOutput] = []
    rejected: list[RejectedRecord] = []

    for record in products:
        try:
            # Validate this single record against ProductInput rules
            product = ProductInput(**record)
            if product.sku in products_db:
                rejected.append(RejectedRecord(
                    record=record,
                    reasons=[f"Duplicate SKU: {product.sku} already exists"],
                ))
                continue
            now = datetime.now().isoformat()
            stored = ProductOutput(
                name=product.name,
                price=product.price,
                category=product.category,
                sku=product.sku,
                ingested_at=now,
            )
            products_db[product.sku] = stored
            last_ingestion_time = now
            accepted.append(stored)

What does this code do?

The function now loops through each record one at a time. Inside the try block, it attempts validation with ProductInput(**record). If validation passes and the SKU is new, the product is stored and added to the accepted list.

If the SKU already exists, the record is added to rejected with a clear reason. The continue statement skips to the next record without crashing the whole batch.

• Directly below the code you just pasted (still inside the for loop), add the exception handler and summary return:

        except Exception as e:
            # Extract human-readable reasons from Pydantic errors
            reasons = []
            if hasattr(e, "errors"):
                for error in e.errors():
                    field = " -> ".join(str(loc) for loc in error["loc"])
                    reasons.append(f"{field}: {error['msg']}")
            else:
                reasons.append(str(e))
            rejected.append(RejectedRecord(record=record, reasons=reasons))

    # Build summary with counts
    summary = {
        "total_submitted": len(products),
        "accepted_count": len(accepted),
        "rejected_count": len(rejected),
    }

    return BatchResponse(accepted=accepted, rejected=rejected, summary=summary)

What does the exception handler do?

When ProductInput(**record) raises a ValidationError, the except block catches it. It checks for a .errors() method (which Pydantic validation errors have) and formats each field error into a readable string like price: Input should be greater than 0.

After the loop finishes, the summary dict gives a quick count of total, accepted, and rejected records. The response includes everything a non-technical stakeholder needs to understand what happened.

• Save main.py. Your dev server should auto-reload.

Seeing an indentation error in the terminal?

The except block must be indented to the same level as the try block (two levels deep: one for the function body, one for the for loop). The summary and return lines sit outside the loop, at one level of indentation.

help me fix an indentation error in my create_products_batch function

Test the resilient batch endpoint

Time to POST the exact same messy NovaMart data that crashed your API in the previous step. Before you run this, think about what you expect to happen now.

• Open http://127.0.0.1:8000/docs in your browser.
• Expand the POST /products/batch endpoint and click Try it out.

Do you think the request will return a 422 error again, or something different? What do you expect to see for the four invalid records?

• Paste the contents of novamart_data.json into the request body field and click Execute.

You now get a 200 response instead of a 422. The JSON body shows three sections:

• accepted contains the four valid products (Wireless Headphones, Running Shoes, Desk Lamp, Water Bottle), now stored in memory.
• rejected contains the four bad records, each with clear English reasons explaining exactly what failed (blank name, negative price, invalid SKU format, zero price).
• summary shows total_submitted: 8, accepted_count: 4, rejected_count: 4.

Why is this better than the 422?

The old endpoint threw away everything when one record was bad. The new endpoint saves what it can and reports what it cannot. A non-technical stakeholder at NovaMart can read the rejection reasons and fix their export template without filing a support ticket.

• Confirm the valid products were actually stored by expanding GET /products, clicking Try it out, then Execute.

You should see the four accepted products returned in clean JSON. The good data survived the messy batch.

Still getting a 422 or seeing an empty response?

If you still see a 422, the old naive function may still be in your file. Make sure you deleted the entire original create_products_batch function and replaced it with the new version.

If you see an empty accepted list, check that the RejectedRecord and BatchResponse models are defined above the endpoint function in your file.

help me debug my batch endpoint that is not returning accepted products

✔️ Awesome, I've got everything!

Great. Double check you have saved your file.

ⓧ I'd like to double check the full code

Scroll up and confirm your main.py contains each of these sections in order:

• Imports (FastAPI, HTTPException, BaseModel, Field, field_validator, datetime).
• The app = FastAPI(...) initialization.
• ProductInput model with the name_not_empty validator.
• ProductOutput model.
• RejectedRecord model (record and reasons fields).
• BatchResponse model (accepted, rejected, summary fields).
• In-memory storage (products_db and last_ingestion_time).
• The read_root endpoint.
• The create_product endpoint (single POST).
• The rewritten create_products_batch endpoint with try/except and BatchResponse return.
• The list_products endpoint.

Your API now handles messy data like a production system. Valid records are never lost because of a neighbour's bad data. Next up, you will add filtering and search so NovaMart's team can query their clean catalog by category and price.

Add Filtering and Complete the Integration

Your batch endpoint now handles messy data gracefully, accepting good records and reporting clear errors for bad ones.

NovaMart's stakeholders want to query their catalog by category and price range. This step completes the integration: data goes in messy, comes out clean and queryable. You now have the full FDE deliverable.

In this step, get ready to:

• Add query parameters to filter products by category and price range.
• Add a SKU lookup endpoint for individual products.
• Add a health check endpoint for system monitoring.

Add query parameters to GET /products

Right now, GET /products returns every product in the database. NovaMart's stakeholders need to slice the catalog by category and price range without downloading everything.

• In main.py, find the import line at the very top of the file:

from fastapi import FastAPI, HTTPException

• Add Query to the import so the line looks like this:

from fastapi import FastAPI, HTTPException, Query

What is Query?

Query is a FastAPI helper that adds validation and documentation to query string parameters. When you declare a parameter with Query(default=None, gt=0), FastAPI automatically validates incoming values and displays them in the interactive docs.

• Now find the list_products function near the bottom of your file:

@app.get("/products", response_model=list[ProductOutput])
def list_products():
    return list(products_db.values())

• Replace the entire list_products function with this version that accepts filter parameters:

@app.get("/products", response_model=list[ProductOutput])
def list_products(
    category: str | None = Query(default=None, description="Filter by category"),
    min_price: float | None = Query(default=None, gt=0, description="Minimum price"),
    max_price: float | None = Query(default=None, gt=0, description="Maximum price"),
):
    # Start with all products
    results = list(products_db.values())
    # Apply each filter only if the parameter was provided
    if category:
        results = [p for p in results if p.category.lower() == category.lower()]
    if min_price is not None:
        results = [p for p in results if p.price >= min_price]
    if max_price is not None:
        results = [p for p in results if p.price <= max_price]
    return results

What does this code do?

Each parameter defaults to None, meaning the filter only applies when the caller passes it. If someone requests /products?category=electronics, only electronics come back.

The gt=0 constraint on price parameters means FastAPI rejects negative or zero prices at the query level before your code even runs.

• Save main.py.
• Switch to your browser and refresh http://127.0.0.1:8000/docs.
• Expand the GET /products panel. You should see three new parameter fields: category, min_price, and max_price.

Don't see the new parameters?

• Make sure you saved the file. The dev server only reloads on save.
• Check that Query is imported from fastapi in the same import line as FastAPI and HTTPException.
• If your terminal shows a syntax error, make sure the function signature's closing parenthesis is followed by a colon.

help me fix my list_products Query parameters

The parameters appear in the docs, but you need data to test them. Re-ingest the NovaMart batch so you have products to filter.

• In the Swagger UI, expand POST /products/batch and click Try it out.
• Open novamart_data.json in your editor, copy the entire JSON array, and paste it into the request body field.
• Click Execute.
• Now expand GET /products, click Try it out, type electronics in the category field, and click Execute.

You should see only the Wireless Headphones product returned (the one valid electronics item from the batch).

Add a SKU lookup endpoint

Stakeholders also need to look up a single product by its unique SKU code.

• In main.py, scroll to the bottom of the file (below the list_products function you just updated) and add this endpoint:

@app.get("/products/{sku}", response_model=ProductOutput)
def get_product(sku: str):
    # Return the product or a clear 404 if it doesn't exist
    if sku not in products_db:
        raise HTTPException(status_code=404, detail=f"Product with SKU {sku} not found")
    return products_db[sku]

What does this code do?

The {sku} in the path tells FastAPI this is a path parameter. When someone requests /products/ELEC-001, FastAPI passes "ELEC-001" as the sku argument.

If the SKU doesn't exist in the database, the endpoint raises a 404 with a human-readable message instead of crashing.

• Save main.py.
• Refresh http://127.0.0.1:8000/docs. You should see the new GET /products/{sku} panel in the docs.
• Click Try it out, enter ELEC-001 in the sku field, and click Execute.

You get a 404 response with the message Product with SKU ELEC-001 not found. That is correct. The server restarted when you saved, which wiped the in-memory database. The endpoint works. You will test it with real data in the final verification below.

Seeing a 422 instead of 404?

• Make sure the get_product route is defined AFTER the list_products route. If you put /products/{sku} before /products, FastAPI may try to validate a literal path segment as a query parameter.
• Confirm the function signature is def get_product(sku: str): with the sku parameter name matching the path template exactly.

help me fix my get_product endpoint

Add a health check endpoint

Production APIs need a health check that monitoring tools can poll. This endpoint reports the system status, how many products are stored, and when the last ingestion happened.

• In main.py, find the BatchResponse model in the Data Models section. Add this new model directly below it:

class HealthResponse(BaseModel):
    status: str
    total_products: int
    last_ingestion: str | None

What does this model do?

HealthResponse defines the shape of the health check response. The last_ingestion field is str | None because when the server first starts, no ingestion has happened yet.

• Now scroll down to the Endpoints section. Find the read_root function and add this endpoint directly below it (before create_product):

@app.get("/health", response_model=HealthResponse)
def health_check():
    # Report current system status
    return HealthResponse(
        status="healthy",
        total_products=len(products_db),
        last_ingestion=last_ingestion_time,
    )

What does this endpoint do?

The health check reads from the same in-memory store as the other endpoints. It tells monitoring tools (or a curious stakeholder) how many products are loaded and when data last arrived.

• Save main.py.
• Refresh http://127.0.0.1:8000/docs and expand the new GET /health panel.
• Click Try it out then Execute.

You should see a response with status: "healthy", total_products: 0, and last_ingestion: null. The server just restarted, so the database is empty.

Seeing an Internal Server Error?

• Make sure HealthResponse is defined in the Data Models section (above the endpoints), not inside or after an endpoint function.
• Check that the health_check function is placed after read_root and before create_product.
• Confirm you added the str | None type hint on last_ingestion. Without it, Pydantic rejects the None value.

help me fix my health check endpoint

Now run the full integration test. All three new features work together once you ingest data.

• In the Swagger UI, expand POST /products/batch and click Try it out.
• Paste the full contents of novamart_data.json into the request body and click Execute.
• Confirm the response shows accepted and rejected products (the same partial-success response from Step 4).

Before you test the next three requests, what do you expect each one to return?

• Test the category filter: expand GET /products, click Try it out, enter electronics in the category field, and click Execute.
• Test the SKU lookup: expand GET /products/{sku}, click Try it out, enter ELEC-001, and click Execute.
• Test the health check: expand GET /health, click Try it out, and click Execute.

Here is what you should see for each:

• GET /products?category=electronics returns only the Wireless Headphones (the single valid electronics product).
• GET /products/ELEC-001 returns the full Wireless Headphones product with its ingested_at timestamp.
• GET /health now shows total_products with the count of accepted products and last_ingestion with a real timestamp.

Why does this matter for FDE work?

You just completed the full integration loop. Data enters messy, gets validated, and comes out clean and queryable. The health check gives operations teams visibility into the system.

This is the deliverable an FDE hands to a customer: a working API with resilient ingestion, useful error messages, and self-reporting health.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and all three endpoints respond correctly in the Swagger UI.

ⓧ I'd like to double check the full code

Your main.py is now too long to show in one block. Scroll through your file and confirm these sections appear in order:

• Imports: from fastapi import FastAPI, HTTPException, Query plus BaseModel, Field, field_validator from pydantic, and datetime.
• Data Models: ProductInput, ProductOutput, RejectedRecord, BatchResponse, then HealthResponse.
• In-Memory Storage: products_db and last_ingestion_time.
• Endpoints in order: read_root, health_check, create_product, create_products_batch, list_products (with Query params), get_product.

Secret mission

Build a Data Quality Dashboard

FDEs don't just fix data problems. They feed patterns back to the product team so problems get fixed at the source. In this mission, you'll build a dashboard endpoint that tracks ingestion statistics over time and generates actionable recommendations for NovaMart's team.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with no paid cloud services. All product data is stored in-memory and disappears the moment you stop the dev server. There are no ongoing costs to worry about.

Decide whether to keep your project files for future reference, pause the running server to free up your terminal, or delete everything entirely.

Resources you used:

• Project folder on your Desktop containing main.py and novamart_data.json.
• Python virtual environment (venv folder inside your project folder).
• FastAPI and its dependencies (installed inside the virtual environment).
• The running FastAPI dev server process.

Keep everything running

No action needed. Choose this if you are still actively building on the API or plan to continue experimenting with new endpoints.

• Your dev server stays running and you can keep testing through http://127.0.0.1:8000/docs.
• Your project files remain on your Desktop, ready to edit.
• Since everything is local and in-memory, there is no cost to keeping this running.

Pause - I'll come back to this later

Stop the running server to free up your terminal. Your project files stay on disk so you can restart later.

• Switch to the terminal where your dev server is running.
• Press Ctrl+C to stop the server.
• Your project files (main.py, novamart_data.json) remain saved on your Desktop.
• To restart later, open a terminal in your project folder, activate your virtual environment, and run fastapi dev main.py again.

Delete - I don't want to use this again

Remove all project files and the virtual environment from your machine. This frees up disk space and removes everything you created.

• Switch to the terminal where your dev server is running.
• Press Ctrl+C to stop the server.
• Deactivate your virtual environment by running:

deactivate

• Navigate to your Desktop and delete the project folder.

On macOS/Linux:

cd ~/Desktop
rm -rf novamart-api

On Windows:

cd ~\Desktop
rmdir /s /q novamart-api

• This removes main.py, novamart_data.json, and the entire venv folder (including FastAPI and all installed packages).
• Python itself remains installed on your machine. It was already there before this project.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade data integration API that ingests messy customer data and serves it clean through validated, queryable endpoints.

You've learned how to:

• Build a REST API from scratch using FastAPI and Pydantic with strict type validation and auto-generated interactive docs.
• Implement the full FDE workflow: scope a customer problem, build the integration, hit production data failures, and ship a resilient fix.
• Create a partial-success ingestion pattern that accepts valid records while rejecting bad ones with clear, actionable error reports for non-technical stakeholders.
• Secret Mission: Build a data quality dashboard that turns ingestion failures into actionable insights for stakeholders.

Ready to quiz yourself?