Python for AI Engineers

◆ Chapter 15

Capstone II: the Customer Feedback Analyzer (Streamlit → FastAPI → Gemini → SQLite)

The full build: a Customer Feedback Analyzer wiring Streamlit, FastAPI, Gemini and SQLite into one working end-to-end app.

~2,790 words · chapter 15 of 15

Chapter 13 had you read a real agent repo and build a loop. That was inward-facing engineering. This capstone is the opposite: a product. Something you could show a hiring manager and say "I built this, here's the URL." We are going to wire a full-stack LLM app end to end — a web UI, a back-end API, a call to a large language model (an LLM: the kind of model that reads and writes text), and a database — and every piece will be small enough to hold in your head.

What we are building

A Customer Feedback Analyzer. A business owner pastes a stack of reviews — one per line — clicks Analyze, and gets back a table. Each review is tagged with:

  • a label: positive, negative, or neutral
  • a score: a number from 0.0 to 1.0
  • a theme: what the review is about — delivery, price, service, food quality, and so on

Above the table we show aggregates: how many reviews, the average score, the percentage that were positive. A Save to database button stores the run. A Load history view brings old runs back.

The architecture

Three tiers, three files. Here is the whole system in words:

Browser (Streamlit: app.py)
      |
      |  HTTP POST /analyze   { "text": "the pizza was cold" }
      v
Back end (FastAPI: api.py)
      |
      |  Gemini SDK call with a JSON schema
      v
Google Gemini  --->  { "label": "negative", "score": 0.2, "theme": "food quality" }
      |
      v
Back to Streamlit  ---> shown in a table
      |
      |  Save
      v
SQLite (database.py)  --->  feedback table

The front end (app.py) is what the user sees. It never talks to Gemini directly. It only ever calls our back end. The back end (api.py) is the only code that knows how to reach Gemini, and it is the only code that holds the API key.

Why split them at all? Streamlit alone could call Gemini. Three reasons we don't:

  1. The key stays server-side. A Streamlit app is a web page. Anything the page can see, a curious visitor can see. If the LLM key lived in app.py, publishing the app would leak it.
  2. The API is reusable. Once /analyze exists, a mobile app, a cron job, or a colleague's script can all use it. The logic isn't trapped inside one UI.
  3. The API is testable on its own. You can hit /analyze with a test tool and check its output without ever opening a browser.

Reading & judging: When you review an LLM app, the first question is where does the API key live? It belongs only in the back end — never in front-end code that gets shipped to a public browser. Splitting the UI from the API is exactly what makes the analysis reusable by other clients and independently testable. A repo where the Streamlit file imports the model SDK and holds the key is a red flag.

JS → Python: This is the same shape as a React front end talking to an Express/Nest back end. Streamlit is your React, FastAPI is your Express, and the browser-to-server hop is a fetch. The only twist is that both halves are Python.

The Gemini SDK — the one new library

Chapter 14 taught Streamlit, FastAPI, and SQLite. The genuinely new piece here is Google's Gemini SDK. An SDK — software development kit — is just the official library for talking to a service.

Install it:

pip install google-genai
# or, with uv:
uv add google-genai

Import it. Note the slightly unusual form — you import genai from the google namespace package:

from google import genai

Getting a key. Go to Google AI Studio at aistudio.google.com, sign in, and create an API key. Put it in a file called .env in your project root:

GOOGLE_API_KEY=your-key-goes-here

A .env file is a plain list of NAME=value lines that hold secrets and config. It is not Python — it is loaded at runtime. Install the loader and call it once at the top of your program:

pip install python-dotenv
from dotenv import load_dotenv

load_dotenv()   # reads .env and puts each line into the environment

The Gemini SDK looks for an environment variable named GOOGLE_API_KEY automatically. So once load_dotenv() has run, you never pass the key by hand — the client finds it.

Reading & judging: .env must be listed in .gitignore so it is never committed. A leaked key in git history is a real incident, not a hypothetical. If you clone a repo and see a .env tracked by git with a live key in it, that is a serious bug — flag it and rotate the key.

The basic call. Three lines:

client = genai.Client()
resp = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="Say hello in one short sentence.",
)
print(resp.text)

genai.Client() makes a client object (it reads the key from the environment). generate_content sends a prompt: model picks which Gemini variant to use — gemini-2.5-flash is the fast, cheap one, good for high-volume tagging. contents is your prompt. resp.text is the model's reply as a string.

Structured output — the important part

For a plain chatbot, a string reply is fine. For our analyzer it is a menace. We need label, score, and theme as real fields, not a paragraph we have to pick apart with string surgery. Gemini can guarantee that.

First, describe the shape you want with a Pydantic model. Pydantic (met in Chapter 14) is the library that turns a class into a validated data shape:

from pydantic import BaseModel
from google.genai import types

class Analysis(BaseModel):
    label: str      # positive | negative | neutral
    score: float    # 0.0 - 1.0
    theme: str      # delivery, price, service, ...

Then pass that model to Gemini as the required output schema:

resp = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=f"Analyze this review: {review_text}",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=Analysis,
    ),
)

analysis = resp.parsed        # a typed Analysis object, not a string

Two settings do the work. response_mime_type="application/json" tells Gemini to reply with JSON, not prose. response_schema=Analysis binds that JSON to our Pydantic model, so Gemini must return exactly the three fields with the right types.

The payoff is the last line. resp.parsed is not a string — it is a filled-in Analysis object. You reach into it with analysis.label, analysis.score, analysis.theme. No json.loads. No stripping ```json fences that models love to wrap around output. No guessing whether score came back as "0.2" (text) or 0.2 (number) — Pydantic already coerced and validated it. If Gemini ever returned something malformed, the parse would fail loudly instead of feeding garbage downstream.

Reading & judging — provider portability: Look back at the OpenAI and Anthropic calls in Chapter 12. This Gemini call is the same three moves: (1) make a client, (2) send a message, (3) bind a schema so the output is structured. Only the attribute names differ — client.models.generate_content here versus client.chat.completions.create or client.messages.create there; response_schema here versus response_format or a tool definition there. Once you can spot those three moves, you can read any LLM SDK cold, even one you've never touched. The vendor is a detail; the pattern is the skill.

The back end — api.py

Now the FastAPI service. It has one endpoint. FastAPI (Chapter 14) turns Python functions into HTTP endpoints and uses Pydantic models to validate what comes in and describe what goes out.

from fastapi import FastAPI
from pydantic import BaseModel
from google import genai
from google.genai import types
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()
client = genai.Client()

class Analysis(BaseModel):
    label: str
    score: float
    theme: str

class Review(BaseModel):
    text: str

@app.post("/analyze", response_model=Analysis)
def analyze(review: Review) -> Analysis:
    resp = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=(
            "You are a customer-feedback analyst. "
            "Classify the review. label is positive, negative, or neutral. "
            "score is 0.0 (very negative) to 1.0 (very positive). "
            "theme is a short noun phrase like delivery, price, service, food quality. "
            f"Review: {review.text}"
        ),
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=Analysis,
        ),
    )
    return resp.parsed

Review is the request body — the JSON the caller sends, just { "text": "..." }. FastAPI validates it: if someone posts without a text field, they get a clear 422 error automatically, before our code runs.

@app.post("/analyze", ...) registers a POST endpoint. When a request arrives, FastAPI parses the body into a Review object and hands it to analyze. We build a prompt that spells out exactly what each field means — this instruction is what keeps Gemini's labels consistent across thousands of reviews. We call Gemini with the structured config and return resp.parsed.

Notice the same Analysis model does double duty: it is the LLM output schema (via response_schema) and the API's response_model. One class defines the contract in two places, so they can never drift apart. That is the kind of tidy reuse Pydantic makes cheap.

Run it:

fastapi dev api.py     # development: auto-reloads on save
# or
fastapi run api.py     # production-ish: no reload

It serves on 127.0.0.1:8000 (that address means "this machine only"). Open http://127.0.0.1:8000/docs in a browser and FastAPI gives you an interactive test page — you can type a review and fire /analyze without writing any client code. Use it to confirm the back end works before touching the front end.

The front end — app.py

Streamlit (Chapter 14) turns a plain Python script into a web app: every widget is a function call, and the script re-runs top to bottom whenever the user interacts.

import streamlit as st
import requests
import database

database.init_db()

st.title("Customer Feedback Analyzer")

raw = st.text_area("Paste reviews, one per line:")

if st.button("Analyze"):
    reviews = [line.strip() for line in raw.splitlines() if line.strip()]
    results = []
    for review in reviews:
        resp = requests.post(
            "http://127.0.0.1:8000/analyze",
            json={"text": review},
        )
        data = resp.json()
        data["review"] = review
        results.append(data)

    st.session_state["results"] = results
    st.dataframe(results)

    count = len(results)
    avg = sum(r["score"] for r in results) / count
    pct_pos = 100 * sum(1 for r in results if r["label"] == "positive") / count
    st.metric("Reviews", count)
    st.metric("Average score", f"{avg:.2f}")
    st.metric("% positive", f"{pct_pos:.0f}%")

The text_area gives a multi-line box. On Analyze we split it: raw.splitlines() breaks the text into lines, .strip() trims whitespace, and if line.strip() drops blank lines. For each review we requests.post to our back end — json={"text": review} sends the body Gemini's endpoint expects — and resp.json() turns the reply into a dict. We stash the original review text alongside its analysis, collect everything into results, and hand the list straight to st.dataframe, which renders a sortable table.

The aggregates are three plain Python one-liners over the list, shown with st.metric (a big-number widget). {avg:.2f} formats to two decimals; {pct_pos:.0f}% rounds to a whole percent.

We saved the results in st.session_state — Streamlit's per-user memory — so the Save and Load buttons can reach them after the script re-runs:

if st.button("Save to database") and "results" in st.session_state:
    database.save(st.session_state["results"])
    st.success("Saved.")

if st.button("Load history"):
    st.dataframe(database.load_history())

JS → Python: That for review in reviews: requests.post(...) loop is exactly a React component fetch-ing an API for each item — just written synchronously and in Python. There's no await, no .then(); requests.post simply blocks until the response lands, then execution continues. Same idea, calmer syntax.

Persistence — database.py

Chapter 14 taught the SQLite mechanics, so here we just show the three functions this app actually uses. SQLite is a database that lives in a single file — no server to run.

import sqlite3

DB = "feedback.db"

def init_db() -> None:
    with sqlite3.connect(DB) as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS feedback (
                id INTEGER PRIMARY KEY,
                review TEXT,
                label TEXT,
                score REAL,
                theme TEXT
            )
            """
        )

def save(results: list[dict]) -> None:
    with sqlite3.connect(DB) as conn:
        conn.executemany(
            "INSERT INTO feedback (review, label, score, theme) VALUES (?, ?, ?, ?)",
            [(r["review"], r["label"], r["score"], r["theme"]) for r in results],
        )

def load_history() -> list[dict]:
    with sqlite3.connect(DB) as conn:
        conn.row_factory = sqlite3.Row
        rows = conn.execute(
            "SELECT review, label, score, theme FROM feedback ORDER BY id DESC"
        ).fetchall()
        return [dict(row) for row in rows]

init_db creates the feedback table once; CREATE TABLE IF NOT EXISTS makes it safe to call on every startup. save uses executemany to insert a whole batch in one go — the ? placeholders are parameterised queries, which prevent SQL injection (never build SQL by string-concatenating user text). load_history sets row_factory = sqlite3.Row so each row behaves like a dict, then returns the rows newest-first. The with sqlite3.connect(...) block commits and closes the connection for you.

Wiring and running the whole thing

Set up the project with uv (Chapter 14's package manager):

uv init feedback-analyzer
cd feedback-analyzer
uv add google-genai fastapi streamlit requests python-dotenv pydantic
uv sync

uv add records each dependency; uv sync installs them into the project's virtual environment. Drop your .env file in, and add .env to .gitignore.

Then run the two processes in two terminals, back end first:

# terminal 1
fastapi run api.py

# terminal 2
streamlit run app.py

Both run at once because they are two separate programs: FastAPI is the always-listening server, Streamlit is the UI that calls it. Kill the API and the Analyze button will error — that dependency is the whole point of the split.

Ship it

You now have a real thing. Finish it like a portfolio piece: write a README.md that says what it does, how to set the key, and the two run commands. Push it to GitHub (with .env ignored). Write a short LinkedIn post — "I built a full-stack LLM app: Streamlit UI, FastAPI back end, Gemini with structured JSON output, SQLite persistence" — and link the repo.

Reading & judging: A shippable project that wires UI + API + LLM + DB tells a reviewer far more than a notebook of clever snippets. It proves you understand where the key lives, how structured output removes fragile parsing, why the tiers are split, and how the pieces run together. That end-to-end judgment — not any single API call — is what "GenAI engineer" actually means on the job.


APPENDIX — ONE-PAGE PYTHON-FOR-GENAI CHEAT SHEET

JS → Python quick map

JavaScript / TypeScript Python
let/const x = 1 x = 1 (no keyword)
null / undefined None (only one)
=== / identity == (no coercion) / is (identity; x is None)
a ? b : c b if a else c
arr.map/filter [f(x) for x in xs if cond] (comprehension)
{...a, ...b} {**a, **b}
arr / obj+Map / Set list / dict / set
for (const x of arr) for x in arr
async/await, Promise.all async/await, asyncio.gather
function*+yield, for await…of generators + yield, async for
TS interface / type type hints + dataclass / TypedDict
zod schema Pydantic BaseModel (runtime validation)
npm/package.json/node_modules uv/pyproject.toml/.venv
ESLint + Prettier / Jest / dotenv ruff / pytest / pydantic-settings
fetch/axios httpx / requests

The idioms that mark fluent Python

  • if items: not if len(items) > 0 — truthiness (empty list/dict/str/0/None are falsy).
  • for i, x in enumerate(xs) / for a, b in zip(as, bs) — never range(len(...)).
  • d.get(k, default) not d[k] when a key may be missing (esp. parsing LLM/JSON output).
  • x is None, f-strings, pathlib.Path, with for every file/client, unpacking (a, *rest = xs).
  • EAFP: try/except over pre-checking. Return new objects; don't mutate inputs.
  • @dataclass(frozen=True) for trusted internal data; Pydantic at every untrusted boundary.

Top-10 red flags (the 10-second smell test)

  1. Hardcoded API key / secret in the diff → block immediately.
  2. Bare except: / except Exception: pass swallowing errors.
  3. Mutable default argument (def f(x=[]) / ={}).
  4. LLM/JSON output used without validation or a try (the #1 GenAI bug).
  5. print() for logging in service/library code.
  6. No timeout and no retry-with-backoff on API/HTTP calls.
  7. Un-awaited coroutine, or blocking I/O (time.sleep, sync HTTP) inside async.
  8. Sequential awaits in a loop that should be asyncio.gathered; or unbounded concurrency with no semaphore (rate-limit bomb).
  9. Real API hit in a unit test instead of a mock (slow, costly, flaky).
  10. Untrusted tool output / retrieved docs treated as trusted instructions (prompt-injection surface); destructive tools with no approval gate.

(Full checklist: Chapter 10. Codebase-reading procedure: Chapter 13.)

You can read it now. Go read some.

◆ You made it to the end

You can now read, call, and ship with LLMs in Python.

That was the whole point — hands on the keyboard, not just theory. This course is free and always will be. If it helped, and you later need someone to take GenAI from pilot to production safely in a regulated setting, that’s the work I do.