ORMs, query builders and migration tools

orm/lazy_loading.py

import sqlite3

from mart import build
from tinyorm import Session

conn = sqlite3.connect(":memory:")
build(conn)

# Code that reads like plain Python: each customer of Pune and their orders.
session = Session(conn, echo=True)
customers = session.customers("Pune")
for customer in customers:
    print(f"   {customer.name}: orders {[o.order_id for o in customer.orders]}")
print(f"{len(session.log)} statements for {len(customers)} customers: 1 for the list, then 1 per customer")

SQL> SELECT customer_id, name FROM customers WHERE city = ? ORDER BY name ['Pune']
SQL> SELECT order_id, status FROM orders WHERE customer_id = ? ORDER BY order_id [1]
   Asha Kulkarni: orders [101, 103]
SQL> SELECT order_id, status FROM orders WHERE customer_id = ? ORDER BY order_id [5]
   Farah Qureshi: orders [105, 107]
SQL> SELECT order_id, status FROM orders WHERE customer_id = ? ORDER BY order_id [7]
   Lakshmi Iyer: orders [106, 110]
4 statements for 3 customers: 1 for the list, then 1 per customer

orm/tinyorm.py

"""A tiny object mapper, enough to see what an ORM sends: rows become objects, and related rows load on first use.

Real ORMs (SQLAlchemy, Django's ORM, Hibernate) do far more, but they load relationships in the same three ways:
lazily when the attribute is first read (the usual default), with a second SELECT and an IN list, or with a JOIN.
"""
import sqlite3


class Session:
    """Runs and logs every statement, as an ORM's SQL echo does, and counts the rows it reads."""

    def __init__(self, conn, echo=False, lazy="select"):
        self.conn, self.echo, self.lazy, self.log, self.rows = conn, echo, lazy, [], 0
        conn.row_factory = sqlite3.Row

    def run(self, sql, params=()):
        text = " ".join(sql.split())
        self.log.append(text)
        if self.echo:
            print("SQL>", text, list(params))
        rows = self.conn.execute(sql, params).fetchall()
        self.rows += len(rows)
        return rows

    def customers(self, city, load=None):
        """Customers of a city, with their orders loaded lazily, by IN list ("selectin") or by JOIN ("joined")."""
        if load == "joined":
            rows = self.run(
                "SELECT c.customer_id, c.name, o.order_id, o.status FROM customers c "
                "LEFT JOIN orders o ON o.customer_id = c.customer_id WHERE c.city = ? ORDER BY c.name, o.order_id",
                (city,),
            )
            found = {}
            for r in rows:
                c = found.setdefault(r["customer_id"], Customer(self, r, orders=[]))
                if r["order_id"] is not None:
                    c._orders.append(Order(r))
            return list(found.values())
        found = [Customer(self, r) for r in self.run(
            "SELECT customer_id, name FROM customers WHERE city = ? ORDER BY name", (city,))]
        if load == "selectin" and found:
            ids = [c.customer_id for c in found]
            rows = self.run(
                f"SELECT order_id, customer_id, status FROM orders WHERE customer_id IN ({', '.join('?' * len(ids))}) "
                "ORDER BY order_id",
                ids,
            )
            by_id = {c.customer_id: c for c in found}
            for c in found:
                c._orders = []
            for r in rows:
                by_id[r["customer_id"]]._orders.append(Order(r))
        return found


class Customer:
    def __init__(self, session, row, orders=None):
        self._session, self.customer_id, self.name, self._orders = session, row["customer_id"], row["name"], orders

    @property
    def orders(self):
        if self._orders is None:
            if self._session.lazy == "raise":
                raise RuntimeError(f"Customer.orders of {self.name} was not loaded, and lazy loading is turned off")
            self._orders = [Order(r) for r in self._session.run(
                "SELECT order_id, status FROM orders WHERE customer_id = ? ORDER BY order_id", (self.customer_id,))]
        return self._orders


class Order:
    def __init__(self, row):
        self.order_id, self.status = row["order_id"], row["status"]

orm/mart.py

"""The mart practice shop with a few more customers and orders, built in any sqlite3 connection."""

SCHEMA = """
CREATE TABLE customers (
  customer_id INTEGER PRIMARY KEY,
  name        TEXT    NOT NULL,
  city        TEXT    NOT NULL
);
CREATE TABLE products (
  product_id  INTEGER PRIMARY KEY,
  name        TEXT    NOT NULL,
  category    TEXT    NOT NULL,
  price_paise INTEGER NOT NULL
);
CREATE TABLE orders (
  order_id    INTEGER PRIMARY KEY,
  customer_id INTEGER NOT NULL REFERENCES customers (customer_id),
  status      TEXT    NOT NULL
);
CREATE TABLE order_items (
  order_id    INTEGER NOT NULL REFERENCES orders (order_id),
  product_id  INTEGER NOT NULL REFERENCES products (product_id),
  quantity    INTEGER NOT NULL CHECK (quantity > 0),
  PRIMARY KEY (order_id, product_id)
);
"""

CUSTOMERS = [
    (1, "Asha Kulkarni", "Pune"), (2, "Vikram Nair", "Kochi"), (3, "Meera Rathore", "Jaipur"),
    (4, "Tenzin Dorjee", "Gangtok"), (5, "Farah Qureshi", "Pune"), (6, "Rohan Das", "Kochi"),
    (7, "Lakshmi Iyer", "Pune"), (8, "Arjun Mehta", "Jaipur"),
]
PRODUCTS = [
    (1, "Basmati rice 1 kg", "Grains", 14999), (2, "Toor dal 500 g", "Pulses", 8950),
    (3, "Masala chai 250 g", "Beverages", 12000), (4, "Coconut oil 1 L", "Oils", 21500),
    (5, "Ragi flour 1 kg", "Grains", 6800),
]
ORDERS = [
    (101, 1, "delivered"), (102, 2, "delivered"), (103, 1, "placed"), (104, 3, "cancelled"),
    (105, 5, "delivered"), (106, 7, "delivered"), (107, 5, "delivered"), (108, 6, "placed"),
    (109, 8, "delivered"), (110, 7, "cancelled"),
]
ORDER_ITEMS = [
    (101, 1, 2), (101, 2, 1), (102, 4, 1), (103, 3, 2), (103, 5, 1), (104, 2, 3), (105, 3, 1),
    (105, 4, 2), (106, 1, 1), (107, 5, 3), (108, 2, 2), (109, 1, 1), (109, 3, 1), (110, 4, 1),
]


def build(conn):
    """Creates the four tables and their rows, then commits."""
    conn.executescript(SCHEMA)
    conn.executemany("INSERT INTO customers VALUES (?, ?, ?)", CUSTOMERS)
    conn.executemany("INSERT INTO products VALUES (?, ?, ?, ?)", PRODUCTS)
    conn.executemany("INSERT INTO orders VALUES (?, ?, ?)", ORDERS)
    conn.executemany("INSERT INTO order_items VALUES (?, ?, ?)", ORDER_ITEMS)
    conn.commit()