Object-Relational Mappers: Code

from flask import Flask, render_template, request, redirect, url_for
import database

# remember to $ pip install flask
# remember to $ pip install peewee

database.initialize("pets.db")

app = Flask(__name__)


def error_page(message, status=400):
    # Return a readable message with the supplied HTTP status.
    return message, status, {"Content-Type": "text/plain; charset=utf-8"}


@app.route("/", methods=["GET"])
@app.route("/list", methods=["GET"])
def get_list():
    pets = database.get_pets()
    return render_template("list.html", pets=pets)


@app.route("/create", methods=["GET"])
def get_create():
    return render_template("create.html")


@app.route("/create", methods=["POST"])
def post_create():
    data = dict(request.form)
    name = (data.get("name") or "").strip()
    pet_type = (data.get("type") or "").strip()
    if name == "":
        return error_page("Error: name is required.", 400)
    if pet_type == "":
        return error_page("Error: type is required.", 400)

    database.create_pet(data)
    return redirect(url_for("get_list"))


@app.route("/delete/<id>", methods=["GET"])
def get_delete(id):
    try:
        # Validate id early so ValueError doesn't become a 500.
        int(id)
    except ValueError:
        return error_page("Error: pet id must be an integer.", 400)

    database.delete_pet(id)
    return redirect(url_for("get_list"))


@app.route("/update/<id>", methods=["GET"])
def get_update(id):
    try:
        int(id)
    except ValueError:
        return error_page("Error: pet id must be an integer.", 400)

    data = database.get_pet(id)
    if data is None:
        return error_page("Error: pet not found.", 404)
    return render_template("update.html", data=data)


@app.route("/update/<id>", methods=["POST"])
def post_update(id):
    try:
        int(id)
    except ValueError:
        return error_page("Error: pet id must be an integer.", 400)

    data = dict(request.form)
    name = (data.get("name") or "").strip()
    pet_type = (data.get("type") or "").strip()
    if name == "":
        return error_page("Error: name is required.", 400)
    if pet_type == "":
        return error_page("Error: type is required.", 400)

    if database.get_pet(id) is None:
        return error_page("Error: pet not found.", 404)

    database.update_pet(id, data)
    return redirect(url_for("get_list"))


@app.route("/health", methods=["GET"])
def health():
    try:
        database.get_pets()
        return error_page("ok", 200)
    except Exception as e:
        return error_page(f"Error checking health: {e}", 500)

"""Pet operations implemented with Peewee, returning ordinary dictionaries.

Peewee, Charles Leifer: models, queries, and writes.
https://docs.peewee-orm.com/en/latest/peewee/models.html
https://docs.peewee-orm.com/en/latest/peewee/querying.html
https://docs.peewee-orm.com/en/latest/peewee/writing.html
"""
import os

from peewee import IntegerField, Model, SqliteDatabase, TextField
from playhouse.migrate import SqliteMigrator, migrate

# Bind the file later so the app and tests can use different databases.
# https://docs.peewee-orm.com/en/latest/peewee/database.html
db = SqliteDatabase(None)


class BaseModel(Model):
    # Meta configures the model; it does not declare stored fields.
    class Meta:
        database = db


class Pet(BaseModel):
    # Class attributes describe columns. Peewee supplies an integer id field.
    name = TextField(null=False)
    type = TextField(null=False)
    age = IntegerField(default=0)
    food = TextField(null=True)


def initialize(database_file):
    close_connection()
    db.init(database_file)
    db.connect(reuse_if_open=True)
    db.create_tables([Pet])

    # Preserve rows in the original example when adding the food field.
    # create_tables() creates missing tables; it does not alter existing ones.
    # https://docs.peewee-orm.com/en/latest/peewee/db_tools.html#schema-migrations
    columns = {column.name for column in db.get_columns("pet")}
    if "food" not in columns:
        migrate(SqliteMigrator(db).add_column("pet", "food", TextField(null=True)))


def close_connection():
    if not db.is_closed():
        db.close()


def _normalize_age(value):
    # Keep the example's conversion rule for values received from a form.
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _pet_values(data):
    name = (data.get("name") or "").strip()
    pet_type = (data.get("type") or "").strip()
    if not name:
        raise ValueError("Pet name is required.")
    if not pet_type:
        raise ValueError("Pet type is required.")
    return {
        "name": name,
        "type": pet_type,
        "age": _normalize_age(data.get("age")),
        "food": data.get("food", ""),
    }


def pet_to_dict(pet):
    # Keep model objects inside this layer. Routes and templates use dictionaries.
    return {
        "id": pet.id, "name": pet.name, "type": pet.type,
        "age": pet.age, "food": pet.food,
    }


def get_pets():
    # select() builds a query. Iteration retrieves rows as Pet objects.
    query = Pet.select().order_by(Pet.name, Pet.id)
    return [pet_to_dict(pet) for pet in query]


def get_pet(id):
    # This comparison builds a SQL condition, not an ordinary Python Boolean.
    pet = Pet.get_or_none(Pet.id == int(id))
    if pet is None:
        return None
    return pet_to_dict(pet)


def create_pet(data):
    # create() inserts immediately and returns the saved model instance.
    pet = Pet.create(**_pet_values(data))
    return pet.id


def delete_pet(id):
    # where() selects the record. execute() runs the DELETE statement.
    Pet.delete().where(Pet.id == int(id)).execute()


def update_pet(id, data):
    # update() builds the statement; execute() sends it to SQLite.
    Pet.update(**_pet_values(data)).where(Pet.id == int(id)).execute()


def setup_test_database(db_file="test_pets.db"):
    close_connection()
    try:
        os.remove(db_file)
    except FileNotFoundError:
        pass

    initialize(db_file)

    pets = [
        {"name": "dorothy", "type": "dog", "age": 9, "food": "kibble"},
        {"name": "suzy", "type": "mouse", "age": 9},
        {"name": "casey", "type": "dog", "age": 9},
        {"name": "heidi", "type": "cat", "age": 15},
    ]
    for pet in pets:
        create_pet(pet)

    assert len(get_pets()) == 4


def test_get_pets():
    pets = get_pets()
    assert type(pets) is list
    assert len(pets) >= 1
    assert type(pets[0]) is dict
    for key in ["id", "name", "type", "age", "food"]:
        assert key in pets[0]
    assert type(pets[0]["name"]) is str


def test_create_pet_and_get_pet():
    new_id = create_pet({"name": "walter", "age": "2", "type": "mouse", "food": "seeds"})
    pet = get_pet(new_id)
    assert pet is not None
    assert pet["name"] == "walter"
    assert pet["age"] == 2
    assert pet["type"] == "mouse"
    assert pet["food"] == "seeds"


def test_update_pet():
    new_id = create_pet({"name": "temp", "age": 1, "type": "cat"})
    update_pet(new_id, {"name": "updated", "age": "8", "type": "dog", "food": "kibble"})
    pet = get_pet(new_id)
    assert pet is not None
    assert pet["name"] == "updated"
    assert pet["age"] == 8
    assert pet["type"] == "dog"
    assert pet["food"] == "kibble"


def test_delete_pet():
    new_id = create_pet({"name": "delete_me", "age": 3, "type": "fish"})
    delete_pet(new_id)
    assert get_pet(new_id) is None


if __name__ == "__main__":
    setup_test_database()
    test_get_pets()
    test_create_pet_and_get_pet()
    test_update_pet()
    test_delete_pet()
    close_connection()
    print("done.")

This is a binary data file. It is available in the repository linked below.

This is a binary data file. It is available in the repository linked below.

Flask==3.1.3
peewee==4.0.0

Script started on 2026-02-23 22:59:37+00:00

$ pip install peewee

Collecting peewee
Downloading peewee-4.0.0-py3-none-any.whl.metadata (8.1 kB)
Downloading peewee-4.0.0-py3-none-any.whl (139 kB)
Installing collected packages: peewee
Successfully installed peewee-4.0.0

[notice] A new release of pip is available: 25.3 -> 26.0.1
[notice] To update, run: python3 -m pip install --upgrade pip

$ python3 -m pip install --upgrade pip

Requirement already satisfied: pip (25.3)
Collecting pip
Downloading pip-26.0.1-py3-none-any.whl.metadata (4.7 kB)
Downloading pip-26.0.1-py3-none-any.whl (1.8 MB)
Installing collected packages: pip
Attempting uninstall: pip 25.3
Successfully uninstalled pip-25.3
Successfully installed pip-26.0.1

$ python

Python 3.12.1 on linux

>>> from peewee import *
>>> db = SqliteDatabase("people.db")

>>> class Person(Model):
...     name = CharField()
...     birthday = DateField()
...     class Meta:
...         database = db

>>> class Pet(Model):
...     owner = ForeignKeyField(Person, backref="pets")
...     name = CharField()
...     animal_type = CharField()
...     class Meta:
...         database = db

>>> db.connect()
True

>>> db.create_tables([Person, Pet])

>>> from datetime import date

>>> uncle_bob = Person(name="Bob", birthday=date(1960, 1, 15))
>>> uncle_bob.save()
1

>>> grandma = Person.create(name="Grandma", birthday=date(1935, 3, 1))
>>> herb = Person.create(name="Herb", birthday=date(1950, 5, 5))

>>> bob_kitty = Pet.create(owner=uncle_bob, name="Kitty", animal_type="cat")
>>> herb_fido = Pet.create(owner=herb, name="Fido", animal_type="dog")
>>> herb_mittens = Pet.create(owner=herb, name="Mittens", animal_type="cat")
>>> herb_mittens_jr = Pet.create(owner=herb, name="Mittens Jr", animal_type="cat")

>>> herb_mittens.delete_instance()
1

>>> grandma = Person.get(Person.name == "Grandma L.")
Traceback (most recent call last):
PersonDoesNotExist: <Model: Person> instance matching query does not exist

>>> grandma = Person.get(Person.name == "Grandma")

>>> grandma.name
'Grandma'

>>> grandma.birthday
datetime.date(1935, 3, 1)

>>> for person in Person.select():
...     print(person.name)
...
Bob
Grandma
Herb

>>> people = Person.select()
>>> list(people)
[<Person: 1>, <Person: 2>, <Person: 3>]

>>> [p.name for p in people]
['Bob', 'Grandma', 'Herb']

>>> query = Pet.select().where(Pet.animal_type == "cat")
>>> for pet in query:
...     print(pet.name, pet.owner.name)
...
Kitty Bob
Mittens Jr Herb

>>> [(pet.name, pet.owner.name) for pet in Pet.select().where(Pet.animal_type == "cat")]
[('Kitty', 'Bob'), ('Mittens Jr', 'Herb')]

"""Prepare the pet table without starting Flask."""
import argparse
import database

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Prepare the Peewee pet database.")
    parser.add_argument("database_file", nargs="?", default="pets.db")
    args = parser.parse_args()
    database.initialize(args.database_file)
    database.close_connection()
    print(f"Ready: {args.database_file}")

<html>

<head></head>

<body>
    This is the create template.
    <form action="/create" method="post">
        <hr />
        <p>Name:<input name="name" /></p>
        <p>Age:<input name="age" /></p>
        <p>Type:<input name="type" /></p>
        <p>Food:<input name="food" /></p>
        <hr />
        <button type="submit">Create</button>
        <a href="/list">Cancel</a>
        <hr />
    </form>
</body>

</html>

<html>
<h3>Pets:</h3>
<table>
    <tr>
        <th>ID</th>
        <th>Name</th>
        <th>Type</th>
        <th>Age</th>
        <th>Food</th>
    </tr>
    {% for pet in pets %}
    <tr>
        <td>{{ pet['id'] }}</td>
        <td>{{ pet['name'] }}</td>
        <td>{{ pet['type'] }}</td>
        <td>{{ pet['age'] }}</td>
        <td>{{ pet['food'] or '' }}</td>
        <td><a href="/delete/{{pet['id']}}">Delete</a></td>
        <td><a href="/update/{{pet['id']}}">Update</a></td>
    </tr>
    {% endfor %}
</table>
<hr />
<a href="/create">Create New Pet</a>

</html>

<html>

<head></head>

<body>
    This is the update template.
    <form action="/update/{{data['id']}}" method="post">
        <hr />
        <p>Name:<input name="name" value="{{data['name']}}" /></p>
        <p>Age:<input name="age" value="{{data['age']}}" /></p>
        <p>Type:<input name="type" value="{{data['type']}}" /></p>
        <p>Food:<input name="food" value="{{data['food'] or ''}}" /></p>
        <hr />
        <button type="submit">Update</button>
        <a href="/list">Cancel</a>
        <hr />
    </form>
</body>

</html>

"""Run with python3 -m unittest -v. All writes use temporary databases."""
from pathlib import Path
import sqlite3
import tempfile
import unittest

import database


class DatabaseTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.path = str(Path(self.temp.name) / 'pets.db')
        database.initialize(self.path)

    def tearDown(self):
        database.close_connection()
        self.temp.cleanup()

    def pet(self):
        return dict(name="O'Malley", type='cat', age='3', food='salmon')

    def test_crud_and_reopen(self):
        pet_id = database.create_pet(self.pet())
        self.assertEqual(database.get_pet(pet_id)['food'], 'salmon')
        database.update_pet(pet_id, {'name': "O'Malley", 'type': 'cat', 'age': '4', 'food': 'tuna'})
        database.close_connection()
        database.initialize(self.path)
        saved = database.get_pet(pet_id)
        self.assertEqual(saved['age'], 4)
        self.assertEqual(saved['food'], 'tuna')
        # Independent SQLite read confirms that the ORM saved a real table row.
        with sqlite3.connect(self.path) as connection:
            self.assertEqual(connection.execute('select food from pet where id=?', (pet_id,)).fetchone(), ('tuna',))
        database.delete_pet(pet_id)
        self.assertIsNone(database.get_pet(pet_id))

    def test_queries_and_validation(self):
        for name in ['Zelda', 'Alex', 'Alex']:
            database.create_pet({'name': name, 'type': 'dog', 'age': 'unknown'})
        pets = database.get_pets()
        self.assertEqual([p['name'] for p in pets], ['Alex', 'Alex', 'Zelda'])
        self.assertEqual(pets[0]['age'], 0)
        self.assertLess(pets[0]['id'], pets[1]['id'])
        for values in [{'name': ' ', 'type': 'dog'}, {'name': 'Casey', 'type': ''}]:
            with self.assertRaises(ValueError):
                database.create_pet(values)
        self.assertEqual(len(database.get_pets()), 3)

    def test_unsaved_object_and_explicit_save(self):
        pet = database.Pet(name='Casey', type='dog', food='kibble')
        self.assertEqual(database.get_pets(), [])
        pet.save()
        pet.food = 'chicken'
        self.assertEqual(database.get_pet(pet.id)['food'], 'kibble')
        pet.save()
        self.assertEqual(database.get_pet(pet.id)['food'], 'chicken')

    def test_atomic_rollback(self):
        with self.assertRaises(ValueError):
            with database.db.atomic():
                database.create_pet(self.pet())
                raise ValueError('Cancel the group of writes')
        self.assertEqual(database.get_pets(), [])

    def test_existing_table_gains_food_without_losing_rows(self):
        old_path = str(Path(self.temp.name) / 'older.db')
        with sqlite3.connect(old_path) as connection:
            connection.execute('create table pet (id integer primary key, name text not null, type text not null, age integer not null)')
            connection.execute("insert into pet values (1, 'Casey', 'dog', 9)")
        database.initialize(old_path)
        self.assertEqual(database.get_pet(1)['name'], 'Casey')
        self.assertIsNone(database.get_pet(1)['food'])
        database.initialize(old_path)
        self.assertEqual(len(database.get_pets()), 1)


if __name__ == '__main__':
    unittest.main()