DEV Community: Chris Kechagias

Building a Chatbot API From Scratch — Part 2: Streaming, Prompt Engineering and Docker

Chris Kechagias — Mon, 06 Apr 2026 05:55:08 +0000

Part 4 actually of building a retail inventory API and then giving it a brain.

In Part 3 I built the chatbot foundation: FastAPI, PostgreSQL, conversation memory, context trimming, rolling summarization, and 13 PRs worth of broken things. The API worked. It remembered what you said. It didn't fall over when the context got too long.

That was enough to call it functional. But it didn't feel finished. No streaming. No real identity. No way to run it anywhere except my machine.

Five PRs later, all of that changed. Some of it was clean. Some of it was not.

PR 14 — Auto-Title Generation

Small PR. Big quality-of-life improvement.

Every new conversation started with the title "New Chat..." and stayed that way forever. I wanted it to generate automatically from the first message, without blocking the response.

The approach: fire a background task after the conversation is created.

if not request.title: ""
    asyncio.create_task(
        update_conversation_title(engine, conversation.id, request.user_message)
    )

asyncio.create_task() schedules it and moves on. The 201 Created fires immediately. The title shows up a second or two later. Clean.

But before I got there, I spent an embarrassing amount of time debugging. The first version of generate_conversation_title was calling the main model (gpt-5-mini) and getting back empty responses. Latency was around 22 seconds. 22 seconds for a title.

The problem was max_completion_tokens. I had it set to 1000 which is too low for reasoning models (they need token budget to think before responding). But even after bumping it, the main model was overkill for something this simple.

The fix was a dual model setup. A utility model (gpt-5-nano) for cheap background tasks, and the main model only for actual chat. After the switch, latency dropped from 22 seconds to under 2. While testing the fix I noticed OpenAI had released gpt-5.4-mini and gpt-5.4-nano in March 2026, so I bumped both models while I was in there. 3x faster, same quality.

latency_ms: 4527  # gpt-5-mini
latency_ms: 1418  # gpt-5.4-mini

The background task lives in summarizer.py and uses that utility model:

async def update_conversation_title(engine, conversation_id, user_message: str):
    """Background task to generate and set a title for a newly created conversation."""
    try:
        with Session(engine) as session:
            conv = session.get(Conversation, conversation_id)
            if not conv:
                return
            title = await generate_conversation_title(user_message)
            if title: ""
                conv.title = title
                session.add(conv)
                session.commit()
                logger.info(f"Title updated for {conversation_id}: {title}")
    except Exception as e:
        logger.error(f"Title generation failed: {e}")

Two lessons that burned me:

Background tasks need their own DB session. You can't pass the request session in (it gets closed before the task runs). Always create a fresh Session(engine) inside the background function.

@handle_openai_errors cannot be used on background tasks. The decorator wraps exceptions into HTTP responses, which makes no sense in a fire-and-forget context. Plain try/except is the right pattern.

PR 15 — Streaming (SSE)

This one took the most time.

The goal was to replace the blocking endpoint (wait for the full response, return it) with a streaming one. Tokens arrive at the client as they're generated, using Server-Sent Events.

The Service Layer

The streaming function is an async generator. This is where the first real problem appeared.

I tried to decorate it with @handle_openai_errors like everything else:

@handle_openai_errors  # THIS BREAKS IT
async def get_chat_completion_stream(...):
    ...
    async for chunk in response:
        yield chunk

The decorator wraps the function with return await func(...). But func is an async generator — you can't await a generator. It returns a generator object, not a coroutine. The error:

object async_generator can't be used in 'await' expression

The fix: remove the decorator entirely and handle errors inline.

async def get_chat_completion_stream(
    messages: list[dict], model: str | None = None, max_retries: int = 2
):
    """Streams a chat completion response from the OpenAI API."""
    model = model or config.openai_model

    for attempt in range(max_retries + 1):
        try:
            response = await client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True,
                stream_options={"include_usage": True},
                max_completion_tokens=config.openai_max_completion_tokens,
            )
            async for chunk in response:
                yield chunk
            return

        except (openai.APIError, openai.APITimeoutError) as e:
            if attempt < max_retries:
                wait_time = (attempt + 1) * 2
                await asyncio.sleep(wait_time)
                continue
            raise OpenAIServiceException(
                message=f"OpenAI stream error: {str(e)}",
                status_code=getattr(e, "status_code", 502),
                error_code="OPENAI_STREAM_ERROR",
            )

Note the stream_options={"include_usage": True}. This tells OpenAI to include token usage in the final chunk, so you don't need tiktoken on the streaming path.

The final chunk has chunk.choices == [] and chunk.usage populated:

async for chunk in get_chat_completion_stream(trimmed_messages):
    if chunk.choices and chunk.choices[0].delta.content:
        delta = chunk.choices[0].delta.content
        full_content += delta
        yield f"data: {json.dumps({'content': delta})}\n\n"

    if chunk.usage:
        metadata = {
            "model": chunk.model,
            "tokens": chunk.usage.total_tokens,
        }

The Controller

The controller returns a StreamingResponse wrapping an async generator. DB writes happen after the stream completes (never during).

async def stream_generator():
    full_content = ""
    metadata = {}
    start_time = time.perf_counter()

    try:
        async for chunk in get_chat_completion_stream(trimmed_messages):
            if chunk.choices and chunk.choices[0].delta.content:
                delta = chunk.choices[0].delta.content
                full_content += delta
                yield f"data: {json.dumps({'content': delta})}\n\n"
            if chunk.usage:
                metadata = {"model": chunk.model, "tokens": chunk.usage.total_tokens}

        # Stream done — now persist
        latency_ms = (time.perf_counter() - start_time) * 1000
        new_msg = Message(
            conversation_id=conversation.id,
            user_message=request.user_message,
            ai_response=full_content,
            ai_model=metadata.get("model", config.openai_model),
            tokens_used=metadata.get("tokens", 0),
            latency_ms=latency_ms,
        )
        db.add(new_msg)
        db.commit()

        yield "data: [DONE]\n\n"

    except Exception as e:
        logger.error(f"Mid-stream failure: {e}")
        yield f"data: {json.dumps({'error': 'Stream interrupted'})}\n\n"

return StreamingResponse(stream_generator(), media_type="text/event-stream")

Use time.perf_counter() for latency, not time.time(). It's higher resolution and not affected by system clock changes.

To verify streaming is actually working (not just dumping everything at once):

curl -N -X POST http://localhost:8000/chat/ \
  -H "Content-Type: application/json" \
  -d '{"user_id": "your-uuid", "user_message": "Tell me a short story"}'

The -N flag disables buffering. If it works, you'll see chunks appear one by one in the terminal.

PR 16 — Composable Prompt System

This is the one I'm most proud of.

The system prompt was hardcoded in config. One string. Not flexible, not maintainable, and impossible to tune without touching code. I wanted something I could compose and experiment with.

The Architecture

The idea: YAML as a control layer, markdown files as the actual content. Each prompt is assembled at request time by layering components in a fixed order:

base → core → styles → rules → intensity

The folder structure:

prompts/
├── prompts.yaml          # which components each prompt uses
├── base/                 # the foundation (default, concise, tutor...)
├── core/                 # identity and persona
├── styles/               # tone modifiers (casual, formal, sarcastic)
├── rules/                # behavioral constraints (communication, factuality...)
└── intensity/            # tone calibration (low, medium, high)

The YAML defines each named prompt:

stoic:
  base: base/default.md
  core: [identity, persona]
  rules: [communication, factuality]
  intensity: medium

summarizer:
  base: base/summarizer.md
  core: [identity]
  rules: [suppression, factuality]
  intensity: high

The PromptLoader

The loader reads everything at startup and caches it in memory. No file I/O on every request.

class PromptLoader:
    def __init__(self):
        self.base_path = Path(__file__).resolve().parent.parent.parent / "prompts"

        with open(self.base_path / "prompts.yaml", "r") as f:
            self.config = yaml.safe_load(f)

        self.cache = {}
        self._preload_files()

    def _preload_files(self):
        for folder in ["base", "core", "rules", "styles", "intensity"]:
            dir_path = self.base_path / folder
            if not dir_path.exists():
                continue
            for file in dir_path.glob("*.md"):
                key = f"{folder}/{file.stem}"
                self.cache[key] = file.read_text().strip()

    def build(self, name: str, intensity_override: str | None = None, **kwargs) -> str:
        cfg = self.config[name]
        parts = []

        base_name = Path(cfg["base"]).stem
        parts.append(self._get(f"base/{base_name}"))

        for section in ["core", "styles", "rules"]:
            for item in cfg.get(section, []):
                parts.append(self._get(f"{section}/{item}"))

        intensity = intensity_override or cfg.get("intensity", "medium")
        parts.append(self._get(f"intensity/{intensity}"))

        prompt = "\n\n".join(p for p in parts if p)
        return prompt.format(**kwargs)

The **kwargs handles variable injection. The summarizer prompt has {input} and {existing_summary} placeholders:

prompt = loader.build(
    "summarizer",
    input=new_content,
    existing_summary=conv.summary or "No previous summary.",
)

One thing to watch: any { or } in a markdown prompt file that isn't a variable placeholder will break prompt.format(**kwargs) with a KeyError. Avoid curly braces in prompt content unless they're intentional variables.

The Persona

The identity prompt is the philosophical foundation of the whole thing. This is what I actually care about:

The assistant exists to sharpen thinking, not replace it.
It does not comfort. It does not flatter. It does not fill silence with noise.
When a question is asked, it answers. When a problem is presented, it cuts to what matters. When thinking is lazy or circular, it names that — once, directly, without judgment.
The goal is not to be helpful in the way a tool is helpful. The goal is to leave the person thinking more clearly than they arrived.

The chat endpoint now accepts a prompt_key field. Omit it and it defaults to "stoic". Every conversation persists the prompt key so it stays consistent across messages.

{
  "user_id": "...",
  "user_message": "What is RAG?",
  "prompt_key": "tutor"
}

The prompt library lives outside app/ as a standalone content layer. It's not application code — it's configuration. Same reasoning as why tests live outside app/.

PR 17 — Post-Feature Cleanup

After three feature PRs, the codebase had accumulated some debt:

chat_controller was still in the file (fully functional, no longer routed anywhere)
Three separate PromptLoader() instances across different files
Missing docstrings on PromptLoader, build(), stream_generator()
No success logging on the streaming path

The singleton fix: export one shared instance from prompt_loader.py and import it everywhere.

# prompt_loader.py — bottom of file
loader = PromptLoader()

# everywhere else
from .prompt_loader import loader

Same pattern as client = AsyncOpenAI(...) in the OpenAI service. One instance, loaded once at startup.

Cleanup PRs don't feel exciting but they're the difference between a codebase you're proud of and one you're embarrassed to show.

PR 18 — Containerization

Dockerfile is deliberately minimal:

FROM python:3.11-slim

WORKDIR /app

COPY pyproject.toml .
COPY uv.lock .

RUN pip install uv
RUN uv sync --frozen --no-dev

COPY . .

EXPOSE 8000

CMD [".venv/bin/uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

uv sync --frozen --no-dev is the important part — reproducible install, no dev dependencies in the image.

The docker-compose.yml spins up the API and a PostgreSQL 15 container, with a health check on the DB before the API starts:

depends_on:
  db:
    condition: service_healthy

Without condition: service_healthy, the API container starts before Postgres is ready and crashes. Learned this the first time.

The .dockerignore uses a whitelist approach (deny everything, explicitly allow what's needed):

*

!pyproject.toml
!uv.lock
!main.py
!app/
!prompts/

The !prompts/ line is easy to miss. The prompt library lives outside app/, so without it the container starts with an empty cache and every loader.build() call returns an empty string. No crash, no error, just silently wrong behavior.

All Docker commands are wired into taskipy:

task build    # docker compose up --build -d
task start    # docker compose up -d
task stop     # docker compose down
task logs     # docker compose logs -f

What's Next

Phase 2 is complete. The API streams responses, has a composable identity, runs in Docker, and manages context intelligently.

Phase 3 starts with a Telegram bot as a real frontend (no more Swagger UI demos). It'll live in a separate repo, store a per-user OpenAI key, and talk to this API over HTTP. After that: testing, then RAG.

The repo is public: GitHub

Follow the journey: LinkedIn | Medium

Built PR by PR. Mistakes included.

Building a Chatbot API From Scratch: 13 PRs, a Lot of Broken Things, and a Context Window That Actually Works

Chris Kechagias — Sat, 28 Mar 2026 12:12:22 +0000

Part 3 of building a retail inventory API and then giving it a brain.

In Part 1 I explained why I archived my first API and started over. In Part 2 I restructured it properly, migrated to Supabase, and got to 97% test coverage.

The retail API is solid now. Working in production. Tests passing. Architecture I can explain.

So I started the next thing: a chatbot API. Same stack, new layer. The goal: a conversational AI service that remembers what you said, manages long conversations intelligently, and eventually connects to the retail inventory data.

This is what the last few weeks looked like.

Why Build the API at All

I could use a ready-made chatbot SDK. Drop in a library, wrap the OpenAI call, done in an afternoon.

The problem is the same one I had with the retail API the first time. I could make it work without understanding any of it.

I wanted to know what happens when a conversation gets too long for the context window. How tokens get counted. Why the response structure changed between model versions. What "streaming" actually means at the transport layer.

The only way to learn that is to build it yourself and break it repeatedly.

The Stack and the Plan

Same core as the retail API: FastAPI, SQLModel, PostgreSQL. Add OpenAI's Python SDK.

Three-layer architecture: routers take requests, controllers handle logic, services talk to external APIs. Models define the data shapes. Everything lives under app/.

The plan was a roadmap of PRs, one feature at a time. No commits directly to main. Every change reviewed before merge.

I had a senior dev reviewing. Same as the retail API. He reviews my PRs, asks hard questions, and doesn't let things slide.

The First Six PRs Were Fast

Scaffold, config, models, logging, error handling, health endpoint. These were mechanical. I'd done the patterns before on the retail API. Took a few days.

The models are worth mentioning because they drove everything else.

Two tables: Conversation (user_id, title, timestamps) and Message (conversation_id, user_message, ai_response, model used, tokens consumed, latency). Every API call gets stored. I wanted to track exactly what the model said, which version said it, how many tokens it used, and how long it took.

This felt like overkill at the time. It wasn't. That data saved me multiple times during debugging.

PR 7: First Working Chat Endpoint

This is where it got real.

The chat endpoint needed to do a few things at once: create or continue a conversation, load the message history, build the right payload for OpenAI, call the API, store the result, return a clean response.

I wrote a single chat_controller function that handles both new and existing conversations. If no conversation_id is in the request, create one. If there is one, fetch the history and continue.

The controller builds the messages array like this:

messages = [{"role": "system", "content": config.openai_system_prompt}]
for msg in history:
    messages.append({"role": "user", "content": msg.user_message})
    messages.append({"role": "assistant", "content": msg.ai_response})
messages.append({"role": "user", "content": request.user_message})

Simple. Obvious in hindsight.

The decorator for OpenAI error handling was more interesting. A @handle_openai_errors wrapper catches APITimeoutError, APIError, and generic exceptions, and converts them into clean HTTP responses with consistent error codes.

The controller doesn't need to know how OpenAI fails( it just calls the service ).

PR 8: The Project Structure Refactor

The retail API taught me about app/ structure. I used it from the start here. But after seven PRs, I had a problem: config, database, and logging were scattered.

PR 8 moved everything infrastructure-related into app/core/. Config lives there. Database engine and session dependency live there. Logger setup lives there. Custom exceptions live there under app/core/errors/.

This sounds like housekeeping. It was. But now when I onboard someone, I can say: "The business logic is in controllers/. The infrastructure is in core/. The data shapes are in models/. Nothing bleeds between them."

That's worth a whole PR.

PR 11: The Debug Session From Hell

I had a working structure. Clean architecture. Good patterns.

Then I ran the API for the first time locally and nothing worked.

The first error was a datetime serialization crash. JSONResponse couldn't serialize a Python datetime object. The fix was one method call — model_dump(mode="json") instead of model_dump(). Mode "json" converts datetimes and UUIDs to strings. Mode "" leaves them as Python objects. I didn't know that distinction existed.

The second error was the OpenAI API key not being found. I'd configured it with pydantic-settings, which reads .env files into a Settings object. What I didn't know: pydantic-settings populates the Settings object, but it doesn't write to os.environ. The OpenAI library reads os.environ. So the key existed in config.openai_api_key but the OpenAI client couldn't see it.

Fix: explicitly pass the key at client initialization.

client = AsyncOpenAI(api_key=config.openai_api_key)

I'd never had to think about this before because I'd never mixed a settings manager with a third-party SDK that reads environment variables directly.

Third error: max_tokens is deprecated for GPT-5 models. Use max_completion_tokens. Got a 400.

Fourth error: temperature isn't supported by GPT-5 mini. Another 400.

Fifth error: responses came back empty. The conversation history was being sent in reverse order ( newest messages first ). The model received the conversation backwards and returned nothing useful. Changed order_by(desc()) to order_by(asc()).

Sixth error: message history serialized as {} in the response. SQLModel's table=True models loaded from the database don't serialize cleanly through Pydantic v2 when using sa_column. The fix: Message.model_validate(msg) on each history item before returning.

Six separate bugs in one session. Each one taught me something I couldn't have read in a tutorial, because tutorials don't show you what breaks when you combine five technologies at once.

The API worked at the end of the day.

Context Window: The Feature I Underestimated

GPT-5 mini has a 128k token context window. Sounds like enough. It isn't a reason to be lazy.

Sending the entire conversation history on every request is wasteful. At scale it's expensive. And for a long conversation, the model spends time processing messages from hours ago that aren't relevant anymore.

The initial implementation was message-count based: keep the last 15 messages. Simple, configurable via env var.

But message count and token count aren't the same thing. Ten short messages is not the same as ten essay-length responses.

After seeing GPT's assessment of the code, I moved to proper token-based trimming using tiktoken. The trimmer preserves the system prompt and the latest user message always, those are non-negotiable. Then it walks backwards through history, removing the oldest pairs of messages until the total token count fits within the limit.

When messages get evicted, they don't disappear. A background task runs after the main response is returned. It takes the evicted messages and calls a cheap utility model to update a rolling summary, which gets stored on the Conversation record and injected back into context on the next request.

The result: conversations can run indefinitely without the model losing the thread of what was discussed early on.

This is the kind of problem that seems solved by "just use a big context window" until you start thinking about cost, latency, and what actually matters in a conversation.

PR 12: CRUD Endpoints and a Database Lesson

PATCH to update a conversation title. DELETE to remove a conversation entirely.

The DELETE looked simple. Fetch the conversation, delete the messages, delete the conversation, commit. Four lines.

It crashed with a foreign key violation. PostgreSQL won't let you delete a conversation record if message records still reference it, even if you've already queued those messages for deletion in the same session. SQLAlchemy batches the deletes and executes them in the wrong order.

The fix is one line: db.flush() after deleting the messages. This forces the message deletions to hit the database before the conversation deletion is issued.

I knew foreign keys existed. I didn't know SQLAlchemy's unit of work could silently reorder operations in a way that breaks FK constraints. Now I do.

What the Code Looks Like Now

The API has:

POST /chat/ — new conversation
POST /chat/{conversation_id} — continue existing conversation
GET /chat/{conversation_id} — full message history
GET /chat/conversations/{user_id} — list user's conversations
PATCH /chat/{conversation_id}/title — rename a conversation
DELETE /chat/{conversation_id} — delete conversation and all messages

Token-based context trimming. Rolling conversation summary. Exponential backoff retry for empty responses. Dual model setup — expensive model for chat, cheap model for background tasks like summarization.

The architecture is clean enough that adding streaming later won't require rewriting the controller ( just changing how the service returns its response ).

What I Learned That I Didn't Expect to Learn

OpenAI changes their API more than I expected. Parameters get deprecated between minor versions. A field that works on GPT-4o doesn't work on GPT-5 mini. If you don't read the changelog you get 400 errors with no obvious explanation.

Background tasks in FastAPI are useful but need their own database sessions. The main request session might be closed by the time the background task runs. Passing the engine and creating a new session inside the task is the right pattern.

TypeVar matters for decorator type safety. @handle_openai_errors originally had Any as the return type, which silently disabled type checking on every wrapped function. Fixing it to Callable[..., Awaitable[T]] took ten minutes and caught nothing immediately, but it makes the codebase defensible.

PR descriptions are not commit messages. A commit says what changed. A PR description explains what you changed, why you changed it, and what wasn't obvious. I'm still getting this wrong, still getting feedback from the senior dev on it, still improving.

Where It Goes From Here

Auto-title generation: the model names the conversation from the first message instead of truncating it to 50 characters.

Streaming responses: instead of waiting for the full reply, tokens arrive as they're generated.

Prompt loader: load system prompts from a file, select them via config. This is where the behavioral prompt work I've been doing separately gets wired in.

After that: Docker, tests, then the RAG layer that connects this chatbot to the retail inventory data.

The retail API is the foundation. This is the layer that makes it conversational. The third layer will make it actually know something about the domain.

Transitioning from retail operations to AI engineering. Building a fashion-focused retail API with a chatbot layer on top.
Follow the journey: GitHub | LinkedIn

Restructuring a FastAPI Project, Migrating to Supabase, and Hitting 97% Test Coverage

Chris Kechagias — Tue, 17 Mar 2026 07:15:59 +0000

Part 2 of building a retail inventory API from scratch.

In Part 1, I explained why I archived my first API and started over. I ended that post with a confession: the new version still had a flat structure and no tests. Not portfolio material yet.

This is what happened next.

The Flat Structure Problem

The v2 codebase worked, but everything lived at the root level. products.py, variants.py, analytics.py, database.py — all siblings, all imports pointing everywhere. When I added a new router, I had to touch three files. When something broke, I had to search across the whole project.

The fix: move everything into an app/ package with proper separation.

app/
├── __init__.py
├── config.py
├── database.py
├── controllers/
├── models/
├── routers/
├── middleware/
└── utils/
    └── errors/

Each folder has a clear responsibility. controllers/ is business logic. routers/ is just route definitions. models/ is data shapes. middleware/ handles logging and exception handling. utils/errors/ defines custom exceptions.

This sounds obvious. It wasn't to me three months ago.

The `init.py` Pattern

Every folder has an __init__.py that re-exports everything inside it. The goal: any file in the project should be able to import from a single clean path instead of hunting through nested modules.

# app/controllers/__init__.py
from .products import (
    create_product_controller as create_product_controller,
    delete_product_controller as delete_product_controller,
    get_product_controller as get_product_controller,
    get_products_controller as get_products_controller,
    update_product_controller as update_product_controller,
)

The as X syntax is not redundant — it tells the linter (ruff) that these re-exports are intentional, not unused imports. Without it, CI fails.

Now a router just does:

from ..controllers import create_product_controller, get_product_controller

Clean. One source of truth per namespace.

Dev Tooling: uv + Taskipy + Ruff

Three tools that changed how I work:

uv — package manager. Replaces pip. Dramatically faster, lock file included, virtual env handled automatically. No more pip install -r requirements.txt and hoping for the best.

Taskipy — task runner. Instead of remembering long commands, I define shortcuts in pyproject.toml:

[tool.taskipy.tasks]
dev = "uvicorn main:app --reload"
build = "docker compose up --build"
lint = "ruff check ."
fix = "ruff check . --fix"
test = "pytest -v"
dev_test = "ruff check . && pytest -v"

task dev_test runs lint then tests in one shot. task fix auto-corrects most ruff violations. Small thing, big quality of life improvement.

Ruff — linter and formatter. Replaces flake8, isort, and black in one tool. Fast, opinionated, catches real problems. Made me write better imports and stop accumulating dead code.

The Bug Safari

Restructuring is not just moving files. Every import path breaks. Here's what I found:

Reserved logging keys. I had this in a controller:

logger.info("Product created", extra={"name": product.name})

This caused a 500 error. name is a reserved key in Python's LogRecord — you can't use it in extra. Renamed to product_name, fixed. The error only appeared at runtime, not at import time.

Route ordering conflict. GET /products/total_value kept returning 422. The analytics router was registered after the products router in main.py. FastAPI matched /total_value as a product ID (a string, hence 422). Moved analytics router first. Fixed.

Missing product_id in variant creation. ProductVariant.model_validate(variant) failed validation because product_id was required but not in the incoming request body — it comes from the URL path. The fix:

db_variant = ProductVariant.model_validate(
    variant, update={"product_id": product_id}
)

Pydantic v2's update parameter merges extra fields into the model during validation.

Migrating to Supabase

The original setup used a local PostgreSQL container via Docker Compose. Fine for development, annoying for deployment. Supabase gives you a managed PostgreSQL instance with a connection pooler.

The config change was minimal — just different environment variables:

# Local Docker
DB_HOST=localhost
DB_PORT=5432

# Supabase (Transaction pooler)
DB_HOST=aws-0-eu-central-1.pooler.supabase.com
DB_PORT=6543

One gotcha: Supabase has two pooler types. Session pooler keeps a persistent connection per client. Transaction pooler reuses connections across requests — better for serverless and low-traffic APIs on free tier. I use the Transaction pooler.

The connection string is assembled in database.py:

engine = create_engine(
    f"postgresql://{config.db_username}:{config.db_password}@{config.db_host}:{config.db_port}/{config.db_name}",
    connect_args={"sslmode": "prefer"},
)

If your password contains special characters (@, #, !), they'll break the URL. Use urllib.parse.quote_plus(password) to encode it.

Writing Tests

This was the part I'd been avoiding. Not because I didn't want tests — because I didn't know how to write them without hitting the real database.

The solution: SQLite in-memory + FastAPI's dependency override pattern.

# tests/conftest.py
from contextlib import asynccontextmanager
from fastapi.testclient import TestClient
import pytest
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from app.database import get_session
from main import app

engine = create_engine(
    "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)

@asynccontextmanager
async def lifespan_override(app):
    yield

@pytest.fixture(name="session")
def session_fixture():
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    SQLModel.metadata.drop_all(engine)

@pytest.fixture(name="client")
def client_fixture(session):
    app.dependency_overrides[get_session] = lambda: session
    app.router.lifespan_context = lifespan_override
    with TestClient(app) as client:
        yield client
    app.dependency_overrides.clear()

Three things happening here:

SQLite in-memory replaces PostgreSQL. No cloud DB needed, no network, instant setup.
get_session dependency is overridden — every request gets the test session, not a real DB connection.
The lifespan is overridden with a no-op — prevents the app from trying to connect to Supabase on test startup.

Each test gets a fresh database. No shared state, no cleanup between tests.

What the Tests Found

This is the part worth highlighting.

While writing tests for the variant endpoints, I noticed the PATCH and DELETE routes didn't validate product existence before operating on variants:

# Before: missing product check
def update_product_variant_router(product_id, variant_id, ...):
    variant = update_product_variant_controller(variant_id, ...)
    if not variant:
        raise ProductVariantNotFoundException(variant_id)
    return variant

You could PATCH /products/9999/variants/1 and it would happily update the variant — even if product 9999 didn't exist. Same for DELETE. Inconsistent behavior, not caught until tests tried to assert PRODUCT_NOT_FOUND.

The fix was two lines per route:

product = get_product_controller(product_id, session)
if not product:
    raise ProductNotFoundException(product_id)

This is exactly what tests are for. Not just confirming the happy path — stress-testing assumptions.

Final result: 29 tests, 97% coverage.

TOTAL    412    11    97%

The 3% uncovered is intentional: database error branches that only trigger on real DB failures, and the Supabase connection code skipped by the lifespan override.

Where It Is Now

The API is live on Render, connected to Supabase, monitored by UptimeRobot.

Live docs: retail-inventory-api-yati.onrender.com/docs

What's still missing: integration tests against the live deployment, an analytics expansion, and a repository pattern refactor I've been putting off.

What's next: a chatbot API. This retail API becomes the data layer. The chatbot queries it.

Transitioning from retail operations to AI engineering. Follow the journey:
GitHub | LinkedIn

Why I Rebuilt My First API From Scratch

Chris Kechagias — Sat, 07 Mar 2026 12:59:28 +0000

I built my first API two months ago. It worked. I deployed it. Users could hit endpoints, get responses, everything functioned. By most measures, it was a success.

Then I looked at the code with someone who actually knew what they were doing, and realized: working isn't the same as professional.

So I archived it and started over.

The First Version

My original retail inventory API did what it was supposed to do. CRUD operations, PostgreSQL database, running in production on Render. I followed tutorials, copied patterns I found online, debugged until things worked.

The problem wasn't that it failed. The problem was I couldn't explain why it succeeded.

When someone asked me to walk through the architecture, I stumbled. Why did I structure the database this way? "That's what the tutorial showed." What's your error handling strategy? I didn't have one—just try-catch blocks scattered everywhere.

I could make it run, but I couldn't defend the design. That bothered me.

Starting Over

The easy path: refactor incrementally. Add tests here, clean up there, gradually improve. But the foundation was shaky. I'd built on patterns I didn't understand. Every new feature would compound that.

Starting from scratch felt wasteful. Two months of work, gone. But those two months taught me what questions to ask. The second time, I wasn't following tutorials blindly—I was making actual decisions.

What Changed

Database Design

Version one: single products table with basic fields. Version two: proper product-variant relationship.

Products have category (Tees, Sweaters, Pants), name, color, price, and collection ("FW24", "SS25"). Variants add size (S-M, L-XL, One Size) with individual quantity and in_stock tracking.

This solves real fashion problems: when a customer wants Size L-XL specifically, you need to know if that variant is in stock, not just whether the base product exists. Categories and sizes use Enums—can't create invalid entries. Field validators catch edge cases like empty strings.

More complex than a single table, but it mirrors how actual retail inventory works.

Architecture

Everything used to live in main.py. Routes, business logic, database queries, all mixed. Worked for five endpoints. Would've been unmaintainable at fifty.

Now: separation of concerns. main.py launches. Services handle logic. Routes define endpoints. Models define data. When I need to change product creation, I know exactly where to look.

Error Handling

Old way: wrap things in try-catch when they broke, handle locally, hope nothing unexpected happens.

New way: centralized error handler. One place catches exceptions, formats responses consistently, logs properly. Something goes wrong anywhere, it gets handled the same way.

Development Workflow

Biggest change wasn't the code—it was how I wrote it. Version one was commits to main whenever something worked. Version two uses pull requests for every feature. Every change gets reviewed before merge.

Felt slow. It is slower. But "slower" means "catching issues before production" and "understanding why patterns matter."

What Code Review Actually Teaches You

Having someone review my code exposed gaps I didn't know existed.

I renamed functions but didn't understand what I was renaming. I was splitting routers from controllers mechanically but couldn't explain the pattern. When asked "are you sure you only need .venv in dockerignore?" I realized I was thinking about exclusions wrong—should've been excluding everything and whitelisting what I need.

I named a docker service web generically. Got told: "You're api. When you add a frontend later, that's web." Right. I was thinking "web server." Should've been thinking "what does this service actually do."

These weren't "change this line" comments. They were "here's why this matters" and "think about scaling." That's what tutorials don't teach.

Some PRs got approved quickly. Others took three rounds. The ones that took longer taught me more.

The Gaps I'm Still Filling

Rebuilding didn't make me an expert. Made me aware of what I don't know.

I implement CRUD operations but couldn't clearly explain in an interview what CRUD means and why it matters as a pattern. I use an ORM (SQLModel) but can't give a solid answer on when to use ORM versus raw SQL and what the tradeoffs are. I have a centralized error handler but had to look up why it's "error handler" not "exception handler" and what that distinction means.

These are basic things. Interview questions. The fact that I'm building working software while having these gaps means I need to go back to fundamentals—not to learn how to code, but to learn how to articulate what I'm doing and why.

The Result

Original API: archived, private. Not bad code—just a snapshot of where I was three months ago. Not portfolio material.

New version isn't production-ready either, still flat, need tests and a proper folder-structure architecture. I'm honest about that. But it's built on patterns I can explain. When I add tests next week, the architecture supports it. When I build a chatbot layer next month, I'll have confidence in the foundation.

If You're In The Same Position

If you built something that works but can't explain it, consider rebuilding. Not because the first version was wrong—because the second version will teach you things the first couldn't.

You'll see patterns you missed. You'll understand why approaches matter. You'll find knowledge gaps you didn't know existed. When someone asks you to explain design decisions, you'll have answers deeper than "the tutorial told me to."

First version proves you can make it work. Second version proves you understand why it works.

That difference matters.

Transitioning from retail operations to AI engineering. Building a fashion-focused retail API with FastAPI + PostgreSQL. Next: chatbot layer, then RAG, then multi-agent systems. Follow the journey: github.com/chris-kechagias/retail-inventory-api