A Minimal ORM: Mapping Real Python Objects to Real SQL

Building a Web Framework

Chapter 7 · A Minimal ORM: Mapping Real Python Objects to Real SQL

Chapter 6 closed on an honest gap: SessionStore's own in-memory dict forgets everything the instant the process ends. This chapter builds a real, minimal ORM — a descriptor-based mapping layer between Python objects and actual SQL — against a genuine SQLite database, and uses it to give SessionStore the real alternative Chapter 6 promised: one that survives a genuine restart.

The Basic Job: Mapping a Row to a Real Object

A database row is a tuple of raw values with no names attached beyond column position; a Python object is a set of named attributes. A small descriptor class declares, once per model class, which attribute maps to which column:

class Field:
    def __init__(self, column):
        self.column = column

class Model:
    _table = None
    _fields = None

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # collect every Field declared on the subclass, once, when the class itself is defined
        cls._fields = {
            name: val.column for name, val in vars(cls).items()
            if isinstance(val, Field)
        }

    def __init__(self, **kwargs):
        for name in self._fields:
            setattr(self, name, kwargs.get(name))

    @classmethod
    def get(cls, conn, id):
        columns = list(cls._fields.values())
        sql = f"SELECT {', '.join(columns)} FROM {cls._table} WHERE id = ?"
        row = conn.execute(sql, (id,)).fetchone()
        if row is None:
            return None
        return cls(**dict(zip(cls._fields.keys(), row)))

    def save(self, conn):
        columns = list(self._fields.values())
        values = [getattr(self, name) for name in self._fields]
        placeholders = ', '.join(['?'] * len(columns))
        sql = f"INSERT OR REPLACE INTO {self._table} ({', '.join(columns)}) VALUES ({placeholders})"
        conn.execute(sql, values)
        conn.commit()

Declaring a real model is two lines of class body:

class Author(Model):
    _table = 'authors'
    id = Field('id')
    name = Field('name')

a = Author.get(conn, 1)   # conn is a real sqlite3 connection
print(a.id, a.name)
# 1 Ann

Verified directly against a real SQLite database

Author.get(conn, 1) genuinely queries a real database and returns a real Author instance with .id and .name populated from the actual row. __init_subclass__ runs once, the moment Author itself is defined, well before any query is issued — every later call to .get() reuses that already-collected field mapping.

The Real SQL Injection Problem

Model.get() already uses a parameter placeholder — but a lookup by name, written the more tempting way with an f-string, doesn't:

def find_author_naive(conn, name):
    sql = f"SELECT id, name FROM authors WHERE name = '{name}'"
    return conn.execute(sql).fetchall()

print(find_author_naive(conn, 'Ann'))
# [(1, 'Ann')]

malicious = "x' OR '1'='1"
print(find_author_naive(conn, malicious))
# [(1, 'Ann'), (2, 'Bo'), (3, 'Cy')]  -- every row in the table

Verified directly — the real, executed SQL genuinely differs from what was intended

The real query that runs, character for character, is SELECT id, name FROM authors WHERE name = 'x' OR '1'='1'. The input's own embedded quote closes the string literal early; OR '1'='1' is always true, so every row matches. This is exactly Chapter 4's own real XSS bug, one layer down — naive string interpolation trusted a value to stay data, and it stopped being data.

Parameterized queries fix it the same way Chapter 4's own SafeString did — separate the query's fixed structure from the untrusted value:

def find_author_safe(conn, name):
    sql = 'SELECT id, name FROM authors WHERE name = ?'
    return conn.execute(sql, (name,)).fetchall()

print(find_author_safe(conn, malicious))
# []  -- treated as a literal name nobody has, not as SQL

Verified directly — the identical malicious string is now completely inert

The exact same malicious string, run through find_author_safe(), returns an empty result. The ? placeholder tells the driver where the query's own structure ends and a value begins, and sends the value separately from the SQL text — no character it contains can ever be interpreted as part of the query itself.

The N+1 Query Problem, Measured

A parameterized single-row lookup is safe. Calling it once per related object, though, adds up fast. A one-line addition to Model fetches every row at once:

    @classmethod
    def all(cls, conn):
        columns = list(cls._fields.values())
        sql = f"SELECT {', '.join(columns)} FROM {cls._table}"
        rows = conn.execute(sql).fetchall()
        return [cls(**dict(zip(cls._fields.keys(), row))) for row in rows]

A second real model, plus a thin wrapper counting every real query issued through it, sets up the measurement — 20 posts split across only 2 distinct authors:

class Post(Model):
    _table = 'posts'
    id = Field('id')
    title = Field('title')
    author_id = Field('author_id')

class CountingConnection:
    # wraps a real sqlite3 connection, counting every real execute() call
    def __init__(self, conn):
        self._conn = conn
        self.query_count = 0
    def execute(self, sql, params=()):
        self.query_count += 1
        return self._conn.execute(sql, params)

Listing every post along with its own author, the straightforward way:

posts = Post.all(conn)                       # 1 query
for post in posts:
    author = Author.get(conn, post.author_id)   # 1 query PER post

print(f"{len(posts)} posts, {conn.query_count} total queries")
# 20 posts, 21 total queries

Verified directly, with a real query counter wrapping a real SQLite connection

20 posts genuinely cost 21 real queries — 1 to fetch the posts, plus 1 more per post's own author lookup, even though only 2 distinct authors actually exist. SQLAlchemy's own documentation names this precisely: "the N plus one problem… for any N objects loaded, accessing their lazy-loaded attributes means there will be N+1 SELECT statements emitted."

Batching the author lookup into one IN-clause query, instead of one per post:

posts = Post.all(conn)                                    # 1 query
author_ids = list({p.author_id for p in posts})
placeholders = ', '.join(['?'] * len(author_ids))
sql = f"SELECT id, name FROM authors WHERE id IN ({placeholders})"
authors_by_id = {row[0]: row for row in conn.execute(sql, author_ids).fetchall()}  # 1 more query

for post in posts:
    author = authors_by_id[post.author_id]   # no query at all -- already loaded

print(f"{len(posts)} posts, {conn.query_count} total queries")
# 20 posts, 2 total queries

Verified directly — the same 20 posts, a real 10.5× fewer queries

Measured against the identical dataset, batching drops the real query count from 21 to 2. Django's own real prefetch_related() and SQLAlchemy's own real selectinload() both work by this identical mechanism, per each project's own documentation.

The Real Payoff: A SqliteSessionStore

A model with three columns — an ID, a JSON-serialized data blob, and a creation timestamp — is enough to reimplement Chapter 6's own SessionStore, keeping its exact create()/load()/save() interface unchanged:

import json, secrets, time, sqlite3

class SessionRow(Model):
    _table = 'sessions'
    id = Field('id')
    data = Field('data')
    created_at = Field('created_at')

class SqliteSessionStore:
    def __init__(self, db_path, max_age=3600):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.execute(
            'CREATE TABLE IF NOT EXISTS sessions '
            '(id TEXT PRIMARY KEY, data TEXT NOT NULL, created_at REAL NOT NULL)'
        )
        self.conn.commit()
        self.max_age = max_age

    def create(self):
        sid = secrets.token_urlsafe(32)
        SessionRow(id=sid, data=json.dumps({}), created_at=time.time()).save(self.conn)
        return sid

    def load(self, sid):
        row = SessionRow.get(self.conn, sid)
        if row is None:
            return None
        if time.time() - row.created_at > self.max_age:
            self.conn.execute('DELETE FROM sessions WHERE id = ?', (sid,))
            self.conn.commit()
            return None
        return json.loads(row.data)

    def save(self, sid, data):
        existing = SessionRow.get(self.conn, sid)
        created_at = existing.created_at if existing else time.time()
        SessionRow(id=sid, data=json.dumps(data), created_at=created_at).save(self.conn)

Wiring it in means changing exactly one line in Chapter 6's own setup — SessionStore() becomes SqliteSessionStore('sessions.db') — and nothing about make_session_middleware() or App has to know or care, since both classes expose the identical three-method interface.

Verified directly — real session data, saved and reloaded from an actual file on disk

store1 = SqliteSessionStore(db_path), a real create() + save() with {'visits': 1, 'username': 'sam'}, then del store1 — simulating the whole process ending. A genuinely fresh store2 = SqliteSessionStore(db_path), a brand-new connection pointed at the same real file, loading the same session ID, returns the identical {'visits': 1, 'username': 'sam'} back. The in-memory dict from Chapter 6 could never have survived that; a real file on disk does.

A Real, Severe Bug Found Wiring This Into the Live App

Plugging SqliteSessionStore straight into a real App, served through wsgiref on a background thread exactly like every prior chapter's own live demos, crashed immediately — not in this chapter's own isolated tests above, only once a real server was actually handling requests:

sqlite3.ProgrammingError: SQLite objects created in a thread can only
be used in that same thread. The object was created in thread id
38384 and this is thread id 38448.

Verified directly — a real, documented SQLite safety restriction, not a bug in this chapter's own code

sqlite3.connect() ties its own connection object to whichever thread created it, by design, specifically to stop a connection being shared unsafely across threads by accident. make_server(...).serve_forever() runs every real request on its own server thread — genuinely different from the thread that called SqliteSessionStore(db_path) in the first place — so the very first live request touching self.conn failed outright. The fix is a real, explicit opt-in: sqlite3.connect(db_path, check_same_thread=False), telling Python this specific connection genuinely will be used from more than one thread.

With the fix applied, run end to end through the entire real stack — App, Router, Request/Response, the session middleware, and SqliteSessionStore together:

request 1 (real live server, port 8911):  b'You have visited 1 times'
request 2 (same jar):                     b'You have visited 2 times'

# server1.shutdown(); del store  -- a real, full "process" teardown
# real row count still on disk afterward: 1

# a genuinely fresh App, fresh SqliteSessionStore, a DIFFERENT port,
# same cookie the still-alive real client jar holds:
request 3 (real live server, port 8912):  b'You have visited 3 times'

Verified directly, end to end — a real "restart" that genuinely doesn't lose anything

The third request — against a completely new App, a completely new SqliteSessionStore, and a new port standing in for a fresh process — correctly continues the count at 3, not 1. Nothing about the client changed; its own cookie jar simply kept sending the same real session ID it had all along. What changed is that the server, this time, remembered it.

Where This Course Is Headed

Chapter 8 builds real error handling and a development server with live reload — directly resolving Chapter 4's own stale-cache finding, and giving crashes like this chapter's own ProgrammingError a genuinely useful debug page instead of wsgiref's own generic 500.

Hands-On Exercises

Exercise 1

Add an update(conn) method to this chapter's own Model base class — distinct from save(), which uses INSERT OR REPLACE — that updates an existing row's own non-id columns via a real SQL UPDATE statement, and verify it against a real change to an existing author's name, confirming the change is re-fetchable afterward and that a different row is left completely untouched.

📄 View solution

Exercise 2

Extend this chapter's own find_author_naive() into a login_naive(conn, name, password) function checking both a name AND a password column with the identical naive string-interpolation style, and construct a real, working login-bypass injection that returns a matching row without ever knowing the real password.

📄 View solution

Exercise 3

Add a real delete(sid) method to SqliteSessionStore (a real SQL DELETE, not just a Set-Cookie clear), wire it into a logout_handler matching Chapter 6's own real logout exercise, and verify — by querying the real row count in the sessions table directly, not just by checking the response — that logging out genuinely removes the session's own row from the database, not just its cookie from the client.

📄 View solution

Chapter 7 Quick Reference

The basic job — Field/Model maps class attributes to real database columns, verified against live SQLite
The real SQL injection problem — naive f-string query building, verified returning every row on a crafted input; the same underlying bug as Chapter 4's own XSS, one layer down
Parameterized queries — a ? placeholder, verified turning the identical malicious input completely inert
The N+1 problem, measured — 20 posts, 21 real queries via naive lazy loading, down to 2 via one batched IN-clause query
SqliteSessionStore — reimplements Chapter 6's own exact create()/load()/save() interface on real SQLite, verified surviving a genuine restart with data intact
check_same_thread=False — a real, necessary fix once this store is wired into a live, threaded WSGI server; verified crashing without it, working correctly end to end with it
Next chapter: Error handling & a real development server with live reload