Learn AI/Projects
LESSON 36 / 36Applied 90 min with practice

Project: a local AI service with a database

Connect a tiny text classifier to a validated HTTP endpoint and SQLite.

WHAT YOU WILL LEARN
  • Expose a local prediction endpoint
  • Validate request size and schema
  • Store results with parameterized SQL

Connect the pieces

An AI application is more than a model. It receives input, validates a contract, performs inference, stores a useful record, and returns a clear result. This project fits a tiny nearest-example text classifier, exposes POST /predict, and records the predicted category in SQLite.

The four training phrases are original fixtures. The classifier uses token-set overlap and returns review when no word matches. It is intentionally modest so the service and data flow stay visible. Replace it with an evaluated model only after the surrounding contract works.

Run locally, then extend deliberately

Save the downloaded file and run python project-local-ai-api-and-database.py. The default mode performs a self-test without opening a server. Add --serve to bind to 127.0.0.1:8000; stop with Ctrl+C. In another terminal, POST JSON such as {"text":"reset password"} to /predict. Results are saved in academy_predictions.sqlite3 in the current folder.

The service restricts request size, checks content type and schema, avoids storing raw text, and uses parameterized SQL. It is a local learning service, not a public deployment template: public use also needs a production server, HTTPS, authentication, authorization, rate limits, and operational monitoring.

PUT THE IDEA INTO CODE

A small experiment you can run.

The database stores only a category and ID. The default self-test avoids starting a long-running process; --serve explicitly enables the local server.

project-local-ai-api-and-database.py
import json, re, sqlite3, sys
from http.server import BaseHTTPRequestHandler, HTTPServer
training = [("reset password login", "account"), ("sign in account", "account"),
            ("invoice payment receipt", "billing"), ("charge billing", "billing")]
def tokens(text):
    return set(re.findall(r"[a-z]+", text.lower()))
def predict(text):
    query = tokens(text)
    scored = [(len(query & tokens(example)), label) for example, label in training]
    score, label = max(scored)
    return label if score else "review"
def validate(payload):
    if not isinstance(payload, dict) or set(payload) != {"text"}:
        raise ValueError("Expected one text field")
    text = payload["text"]
    if not isinstance(text, str) or not 1 <= len(text.strip()) <= 500:
        raise ValueError("Use 1 to 500 characters")
    return text.strip()
class Handler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass  # Do not log request content in this learning service.
    def reply(self, status, data):
        body = json.dumps(data).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)
    def do_POST(self):
        if self.path != "/predict":
            return self.reply(404, {"error": "Not found"})
        if self.headers.get_content_type() != "application/json":
            return self.reply(415, {"error": "Use application/json"})
        try:
            size = int(self.headers.get("Content-Length", "0"))
            if not 0 < size <= 4096:
                return self.reply(413, {"error": "Body must be 1 to 4096 bytes"})
            payload = json.loads(self.rfile.read(size))
            label = predict(validate(payload))
        except (ValueError, UnicodeError):
            return self.reply(400, {"error": "Invalid JSON or text"})
        with sqlite3.connect("academy_predictions.sqlite3") as db:
            db.execute("CREATE TABLE IF NOT EXISTS predictions (id INTEGER PRIMARY KEY, label TEXT NOT NULL)")
            result = db.execute("INSERT INTO predictions (label) VALUES (?)", (label,))
            prediction_id = result.lastrowid
        self.reply(200, {"id": prediction_id, "label": label})
if __name__ == "__main__":
    if "--serve" in sys.argv:
        print("Local learning API: http://127.0.0.1:8000/predict")
        HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()
    else:
        assert predict(validate({"text": "reset password"})) == "account"
        assert predict("galactic weather") == "review"
        print("Self-test passed. Add --serve to run the local API.")
Copy code

Save the file, open your terminal in that folder, and run python project-local-ai-api-and-database.py. Use python3 or py if required by your installation. Setup guide

What to expect

The default mode prints Self-test passed. In server mode, a valid prediction creates one SQLite row.

YOUR TURN

Complete an end-to-end request.

  1. Start the server with --serve and POST a valid JSON request to /predict using your HTTP client.
  2. Send an empty text field and confirm a 400 response; send text/plain and confirm 415.
  3. Inspect the SQLite predictions table and verify invalid requests did not insert rows.
Compare with a suggested solution

A valid request returns an ID and label, such as account. Invalid input returns a structured error before the INSERT. To extend it, add model_version and created_at fields, then test migrations and authorization before any public deployment.

CHECK YOUR UNDERSTANDING

One idea to take with you.

Why use placeholders in the INSERT statement?

Make it part of your progress.

Finish the practice and answer the knowledge check to mark this lesson complete.

Go deeper with primary documentation

Optional references for further study. This lesson and its examples were written for Artificials.

Python SQLite modulePython HTTP server: limitations