ORM with Constraints: Code

from flask import Flask, redirect, render_template, request, url_for
from peewee import IntegrityError, OperationalError

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 message, status, {"Content-Type": "text/plain; charset=utf-8"}


def check_pet_form(data):
    """Return a message for a bad pet form, or None when the form is usable."""
    if (data.get("name") or "").strip() == "":
        return "Error: name is required."
    if (data.get("type") or "").strip() == "":
        return "Error: type is required."
    age = (data.get("age") or "").strip()
    if age != "" and not age.isdigit():
        return "Error: age must be a whole number, zero or more."
    owner_id = (data.get("owner_id") or "").strip()
    if owner_id == "":
        return "Error: You must select an owner for the pet."
    if not owner_id.isdigit():
        return "Error: owner_id must be a number."
    return None


@app.route("/", methods=["GET"])
@app.route("/list", methods=["GET"])
def get_list():
    try:
        pets = database.get_pets()
        return render_template("list.html", pets=pets)
    except OperationalError as e:
        return error_page(f"Database error while listing pets: {e}", 500)


@app.route("/create", methods=["GET"])
def get_create():
    try:
        owners = database.get_owners()
        return render_template("create.html", owners=owners)
    except OperationalError as e:
        return error_page(f"Database error while loading owners: {e}", 500)


@app.route("/create", methods=["POST"])
def post_create():
    data = dict(request.form)

    problem = check_pet_form(data)
    if problem:
        return error_page(problem, 400)

    try:
        database.create_pet(data)
        return redirect(url_for("get_list"))
    except IntegrityError as e:
        return error_page(f"Constraint error creating pet: {e}", 400)
    except OperationalError as e:
        return error_page(f"Database operational error creating pet: {e}", 500)
    except ValueError as e:
        return error_page(f"Bad input creating pet: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error creating pet: {e}", 500)


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

    try:
        database.delete_pet(id)
        return redirect(url_for("get_list"))
    except IntegrityError as e:
        return error_page(f"Constraint error deleting pet: {e}", 400)
    except OperationalError as e:
        return error_page(f"Database operational error deleting pet: {e}", 500)
    except Exception as e:
        return error_page(f"Unexpected error deleting pet: {e}", 500)


@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)

    try:
        data = database.get_pet(id)
        if data is None:
            return error_page("Error: pet not found.", 404)
        owners = database.get_owners()
        return render_template("update.html", data=data, owners=owners)
    except OperationalError as e:
        return error_page(f"Database error loading pet for update: {e}", 500)


@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)

    problem = check_pet_form(data)
    if problem:
        return error_page(problem, 400)

    try:
        database.update_pet(id, data)
        return redirect(url_for("get_list"))
    except IntegrityError as e:
        return error_page(f"Constraint error updating pet: {e}", 400)
    except OperationalError as e:
        return error_page(f"Database operational error updating pet: {e}", 500)
    except ValueError as e:
        return error_page(f"Bad input updating pet: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error updating pet: {e}", 500)


@app.route("/owners", methods=["GET"])
def get_owners_list():
    try:
        owners = database.get_owners()
        return render_template("owner_list.html", owners=owners)
    except OperationalError as e:
        return error_page(f"Database error while listing owners: {e}", 500)


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


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

    try:
        database.create_owner(data)
        return redirect(url_for("get_owners_list"))
    except IntegrityError as e:
        return error_page(f"Constraint error creating owner: {e}", 400)
    except OperationalError as e:
        return error_page(f"Database operational error creating owner: {e}", 500)
    except ValueError as e:
        return error_page(f"Bad input creating owner: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error creating owner: {e}", 500)


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

    try:
        database.delete_owner(id)
        return redirect(url_for("get_owners_list"))
    except IntegrityError as e:
        return error_page(
            "Error: Cannot delete this owner because they have pets. "
            "Please delete their pets first.\n"
            f"(details: {e})",
            400,
        )
    except OperationalError as e:
        return error_page(f"Database operational error deleting owner: {e}", 500)
    except Exception as e:
        return error_page(f"Unexpected error deleting owner: {e}", 500)


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

    try:
        data = database.get_owner(id)
        if data is None:
            return error_page("Error: owner not found.", 404)
        return render_template("owner_update.html", data=data)
    except OperationalError as e:
        return error_page(f"Database error loading owner for update: {e}", 500)


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

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

    try:
        database.update_owner(id, data)
        return redirect(url_for("get_owners_list"))
    except IntegrityError as e:
        return error_page(f"Constraint error updating owner: {e}", 400)
    except OperationalError as e:
        return error_page(f"Database operational error updating owner: {e}", 500)
    except ValueError as e:
        return error_page(f"Bad input updating owner: {e}", 400)
    except Exception as e:
        return error_page(f"Unexpected error updating owner: {e}", 500)


@app.route("/health", methods=["GET"])
def health():
    try:
        fk = database.db.execute_sql("PRAGMA foreign_keys").fetchone()[0]
        if fk != 1:
            return error_page("Error: foreign key constraints are NOT active.", 500)
        return error_page("ok", 200)
    except Exception as e:
        return error_page(f"Error checking health: {e}", 500)

import os
from pathlib import Path
import sqlite3

from peewee import (
    Check,
    ForeignKeyField,
    IntegerField,
    IntegrityError,
    Model,
    SqliteDatabase,
    TextField,
)

# Bind the file later so the app and tests can use different databases.
db = SqliteDatabase(None)


class BaseModel(Model):
    class Meta:
        database = db


class Owner(BaseModel):
    name = TextField(null=False, constraints=[Check("length(trim(name)) > 0")])
    city = TextField(null=True)
    type_of_home = TextField(null=True)


class Pet(BaseModel):
    name = TextField(null=False, constraints=[Check("length(trim(name)) > 0")])
    type = TextField(null=False, constraints=[Check("length(trim(type)) > 0")])
    age = IntegerField(default=0, constraints=[Check("age >= 0")])
    owner = ForeignKeyField(Owner, backref="pets", null=False, on_delete="RESTRICT")


class SchemaNotCurrent(Exception):
    """An existing database file does not have this example's tables and rules."""


def schema_problems(database_file):
    """List the known differences between an existing file and the models.

    This is a heuristic. It recognizes the layouts earlier examples produced
    (no owner table, no owner_id, no ON DELETE RESTRICT, no CHECK rules) and
    does not compare every rule or index. A missing file or a file without
    owner and pet tables has no problems, because create_tables() can build
    what is missing. The file is opened read-only and never changed.
    """
    path = Path(database_file)
    if not path.exists():
        return []
    connection = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
    try:
        sql = dict(connection.execute(
            "select name, sql from sqlite_master "
            "where type = 'table' and name in ('owner', 'pet')").fetchall())
        pet_columns = {row[1] for row in connection.execute("pragma table_info(pet)")}
    finally:
        connection.close()

    problems = []
    if "pet" in sql:
        pet_sql = sql["pet"].lower()
        if "owner" not in sql:
            problems.append("There is a pet table but no owner table.")
        if "owner_id" not in pet_columns:
            problems.append("The pet table has no owner_id column.")
        elif "on delete restrict" not in pet_sql:
            problems.append("The pet table does not restrict deleting an owner who has pets.")
        if "check" not in pet_sql:
            problems.append("The pet table has no CHECK rules.")
    if "owner" in sql and "check" not in sql["owner"].lower():
        problems.append("The owner table has no CHECK rule on name.")
    return problems


def initialize(database_file):
    if not db.is_closed():
        db.close()

    # create_tables() never changes a table that already exists, so an older
    # file has to be rebuilt with upgrade_database.py. Nothing here drops data.
    problems = schema_problems(database_file)
    if problems:
        raise SchemaNotCurrent(
            f"{database_file} does not match this example's tables. "
            + " ".join(problems)
            + " Run python3 upgrade_database.py to build current tables "
            "and reload the data."
        )

    db.init(database_file, pragmas={"foreign_keys": 1})
    db.connect(reuse_if_open=True)
    db.create_tables([Owner, Pet])

    fk = db.execute_sql("PRAGMA foreign_keys").fetchone()[0]
    assert fk == 1, "Foreign key constraints are not active on this connection."


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


def _normalize_age(value):
    if value is None:
        return 0
    if isinstance(value, str) and value.strip() == "":
        return 0

    age = int(value)
    if age < 0:
        raise ValueError("Age must be non-negative.")
    return age


def owner_to_dict(owner):
    return {
        "id": owner.id,
        "name": owner.name,
        "city": owner.city,
        "type_of_home": owner.type_of_home,
    }


def pet_to_dict(pet):
    return {
        "id": pet.id,
        "name": pet.name,
        "type": pet.type,
        "age": pet.age,
        "owner_id": pet.owner_id,
        "owner_name": pet.owner.name,
    }


def get_owners():
    query = Owner.select().order_by(Owner.name, Owner.id)
    return [owner_to_dict(owner) for owner in query]


def get_owner(id):
    owner = Owner.get_or_none(Owner.id == int(id))
    if owner is None:
        return None
    return owner_to_dict(owner)


def create_owner(data):
    name = (data.get("name") or "").strip()
    if name == "":
        raise ValueError("Owner name is required.")

    owner = Owner.create(
        name=name,
        city=(data.get("city") or "").strip() or None,
        type_of_home=(data.get("type_of_home") or "").strip() or None,
    )
    return owner.id


def delete_owner(id):
    owner = Owner.get_or_none(Owner.id == int(id))
    if owner is None:
        return
    owner.delete_instance()


def update_owner(id, data):
    owner = Owner.get_or_none(Owner.id == int(id))
    if owner is None:
        return

    name = (data.get("name") or "").strip()
    if name == "":
        raise ValueError("Owner name is required.")

    owner.name = name
    owner.city = (data.get("city") or "").strip() or None
    owner.type_of_home = (data.get("type_of_home") or "").strip() or None
    owner.save()


def get_pets():
    query = Pet.select(Pet, Owner).join(Owner).order_by(Pet.name, Pet.id)
    return [pet_to_dict(pet) for pet in query]


def get_pet(id):
    pet = (
        Pet.select(Pet, Owner)
        .join(Owner)
        .where(Pet.id == int(id))
        .first()
    )
    if pet is None:
        return None
    return pet_to_dict(pet)


def create_pet(data):
    name = (data.get("name") or "").strip()
    pet_type = (data.get("type") or "").strip()
    owner_id = data.get("owner_id")

    if name == "":
        raise ValueError("Pet name is required.")
    if pet_type == "":
        raise ValueError("Pet type is required.")
    if owner_id is None or str(owner_id).strip() == "":
        raise ValueError("owner_id is required.")

    pet = Pet.create(
        name=name,
        type=pet_type,
        age=_normalize_age(data.get("age")),
        owner=int(owner_id),
    )
    return pet.id


def delete_pet(id):
    Pet.delete().where(Pet.id == int(id)).execute()


def update_pet(id, data):
    pet = Pet.get_or_none(Pet.id == int(id))
    if pet is None:
        return

    name = (data.get("name") or "").strip()
    pet_type = (data.get("type") or "").strip()
    owner_id = data.get("owner_id")

    if name == "":
        raise ValueError("Pet name is required.")
    if pet_type == "":
        raise ValueError("Pet type is required.")
    if owner_id is None or str(owner_id).strip() == "":
        raise ValueError("owner_id is required.")

    pet.name = name
    pet.type = pet_type
    pet.age = _normalize_age(data.get("age"))
    pet.owner = int(owner_id)
    pet.save()


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

    initialize(db_file)

    owners = [
        {"name": "greg", "city": "Portland", "type_of_home": "condo"},
        {"name": "david", "city": "Seattle", "type_of_home": "farm"},
    ]
    owner_ids = {}
    for owner in owners:
        owner_id = create_owner(owner)
        owner_ids[owner["name"]] = owner_id

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

    assert len(get_pets()) == 4
    return owner_ids


def test_constraints_are_active():
    fk = db.execute_sql("PRAGMA foreign_keys").fetchone()[0]
    assert fk == 1


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", "owner_id", "owner_name"]:
        assert key in pets[0]
    assert type(pets[0]["name"]) is str


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


def test_fk_rejects_bad_owner_id():
    try:
        create_pet({"name": "ghost", "age": 1, "type": "dog", "owner_id": 999999})
        assert False, "Expected FOREIGN KEY constraint failure, but insert succeeded."
    except IntegrityError as e:
        msg = str(e).lower()
        assert "foreign key" in msg or "constraint" in msg


def test_delete_owner_restricted(owner_ids):
    try:
        delete_owner(owner_ids["greg"])
        assert False, "Expected delete restriction failure, but delete succeeded."
    except IntegrityError as e:
        msg = str(e).lower()
        assert "foreign key" in msg or "constraint" in msg


def test_delete_pet_then_delete_owner_succeeds():
    owner_id = create_owner({"name": "solo", "city": "Akron", "type_of_home": "house"})
    pet_id = create_pet(
        {"name": "onepet", "age": 3, "type": "cat", "owner_id": owner_id}
    )

    try:
        delete_owner(owner_id)
        assert False, "Expected delete restriction failure, but delete succeeded."
    except IntegrityError:
        pass

    delete_pet(pet_id)
    delete_owner(owner_id)
    assert get_owner(owner_id) is None


def test_get_owners():
    owners = get_owners()
    assert type(owners) is list
    assert len(owners) >= 1
    assert type(owners[0]) is dict
    for key in ["id", "name", "city", "type_of_home"]:
        assert key in owners[0]


def test_get_owner():
    owner = get_owner(2)
    assert owner["name"] == "david"


def test_create_owner():
    create_owner({"name": "santa", "city": "north pole", "type_of_home": "workshop"})
    owners = get_owners()
    owners = [owner for owner in owners if owner["name"] == "santa"]
    owner = owners[0]
    assert owner["name"] == "santa"
    assert owner["city"] == "north pole"
    assert owner["type_of_home"] == "workshop"


def test_update_owner():
    owners = get_owners()
    owner = [owner for owner in owners if owner["name"] == "david"][0]
    owner["name"] = "dave"
    owner["city"] = "riverside"
    owner["type_of_home"] = "suburban"
    owner_id = owner["id"]
    update_owner(owner_id, owner)
    owner = get_owner(owner_id)
    assert owner["id"] == owner_id
    assert owner["name"] == "dave"
    assert owner["city"] == "riverside"
    assert owner["type_of_home"] == "suburban"


def test_delete_owner():
    owners = get_owners()
    owner = [owner for owner in owners if owner["name"] == "santa"][0]
    owner_id = owner["id"]
    delete_owner(owner_id)
    owners = get_owners()
    owners = [owner for owner in owners if owner["name"] == "santa"]
    assert owners == []


if __name__ == "__main__":
    owner_ids = setup_test_database()

    test_constraints_are_active()
    test_get_pets()
    test_create_pet_and_get_pet(owner_ids)
    test_fk_rejects_bad_owner_id()
    test_delete_owner_restricted(owner_ids)
    test_delete_pet_then_delete_owner_succeeds()
    test_get_owners()
    test_get_owner()
    test_create_owner()
    test_update_owner()
    test_delete_owner()
    close_connection()
    print("done.")

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

Flask==3.1.3
peewee==4.0.0

"""Prepare the owner and pet tables without starting Flask."""
import argparse
import database

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Prepare the Peewee pets and owners database.")
    parser.add_argument("database_file", nargs="?", default="pets.db")
    args = parser.parse_args()
    try:
        database.initialize(args.database_file)
    except database.SchemaNotCurrent as error:
        raise SystemExit(str(error))
    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" required /></p>
        <p>Age:<input name="age" type="number" min="0" step="1" /></p>
        <p>Type:<input name="type" required /></p>
        <p>Owner:
            <select name="owner_id" required>
                <option value="">-- Select an Owner --</option>
                {% for owner in owners %}
                <option value="{{ owner['id'] }}">{{ owner['name'] }}</option>
                {% endfor %}
            </select>
        </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>Owner</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['owner_name'] }}</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>
<hr />
<a href="/owners">Manage Owners</a>

</html>

<html>

<head></head>

<body>
    This is the owner create template.
    <form action="/owner/create" method="post">
        <hr />
        <p>Name:<input name="name" required /></p>
        <p>City:<input name="city" /></p>
        <p>Type of Home:<input name="type_of_home" /></p>
        <hr />
        <button type="submit">Create</button>
        <a href="/owners">Cancel</a>
        <hr />
    </form>
</body>

</html>

<html>
<h3>Owners:</h3>
<table>
    <tr>
        <th>ID</th>
        <th>Name</th>
        <th>City</th>
        <th>Type of Home</th>
    </tr>
    {% for owner in owners %}
    <tr>
        <td>{{ owner['id'] }}</td>
        <td>{{ owner['name'] }}</td>
        <td>{{ owner['city'] or '' }}</td>
        <td>{{ owner['type_of_home'] or '' }}</td>
        <td><a href="/owner/delete/{{owner['id']}}">Delete</a></td>
        <td><a href="/owner/update/{{owner['id']}}">Update</a></td>
    </tr>
    {% endfor %}
</table>
<hr />
<a href="/owner/create">Create New Owner</a>
<hr />
<a href="/list">Back to Pets</a>

</html>

<html>

<head></head>

<body>
    This is the owner update template.
    <form action="/owner/update/{{data['id']}}" method="post">
        <hr />
        <p>Name:<input name="name" value="{{data['name']}}" required /></p>
        <p>City:<input name="city" value="{{data['city'] or ''}}" /></p>
        <p>Type of Home:<input name="type_of_home" value="{{data['type_of_home'] or ''}}" /></p>
        <hr />
        <button type="submit">Update</button>
        <a href="/owners">Cancel</a>
        <hr />
    </form>
</body>

</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']}}" required /></p>
        <p>Age:<input name="age" type="number" min="0" step="1" value="{{data['age']}}" /></p>
        <p>Type:<input name="type" value="{{data['type']}}" required /></p>
        <p>Owner:
            <select name="owner_id" required>
                {% for owner in owners %}
                <option value="{{ owner['id'] }}" {% if owner['id']==data['owner_id'] %}selected{% endif %}>{{
                    owner['name'] }}</option>
                {% endfor %}
            </select>
        </p>
        <hr />
        <button type="submit">Update</button>
        <a href="/list">Cancel</a>
        <hr />
    </form>
</body>

</html>

"""Run with python3 -m unittest -v test_app.py. Uses temporary databases."""
import os
from pathlib import Path
import tempfile
import unittest

# app.py opens pets.db when it is imported, so import it from a scratch folder.
SCRATCH = tempfile.TemporaryDirectory()
HERE = os.getcwd()
os.chdir(SCRATCH.name)
import app
import database
os.chdir(HERE)


class WebLayerTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        database.initialize(str(Path(self.temp.name) / "pets.db"))
        self.client = app.app.test_client()
        self.owner_id = database.create_owner({"name": "Alex"})

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

    def form(self, **changes):
        values = {"name": "Casey", "type": "dog", "age": "9", "owner_id": str(self.owner_id)}
        values.update(changes)
        return values

    def test_forms_carry_browser_rules(self):
        page = self.client.get("/create").get_data(as_text=True)
        for text in ['name="name" required', 'name="type" required',
                     'type="number" min="0"', 'name="owner_id" required']:
            self.assertIn(text, page)
        self.assertIn('name="name" required', self.client.get("/owner/create").get_data(as_text=True))

    def test_the_route_rejects_bad_forms_with_a_message(self):
        cases = [(dict(name=" "), "name is required"), (dict(type=""), "type is required"),
                 (dict(age="-1"), "age must be a whole number"), (dict(age="abc"), "age must be a whole number"),
                 (dict(owner_id=""), "must select an owner"), (dict(owner_id="x"), "owner_id must be a number")]
        for changes, message in cases:
            with self.subTest(changes=changes):
                response = self.client.post("/create", data=self.form(**changes))
                self.assertEqual(response.status_code, 400)
                self.assertIn(message, response.get_data(as_text=True))
        self.assertEqual(database.get_pets(), [])

    def test_the_route_stops_bad_updates_too(self):
        pet_id = database.create_pet(self.form())
        response = self.client.post(f"/update/{pet_id}", data=self.form(age="-5"))
        self.assertEqual(response.status_code, 400)
        self.assertEqual(database.get_pet(pet_id)["age"], 9)

    def test_the_database_still_answers_when_the_route_passes_a_bad_reference(self):
        response = self.client.post("/create", data=self.form(owner_id="999999"))
        self.assertEqual(response.status_code, 400)
        self.assertIn("Constraint error", response.get_data(as_text=True))
        self.assertEqual(database.get_pets(), [])

    def test_a_good_form_is_saved_and_a_blank_age_is_zero(self):
        self.assertEqual(self.client.post("/create", data=self.form(age="")).status_code, 302)
        self.assertEqual(database.get_pets()[0]["age"], 0)

    def test_owner_with_pets_cannot_be_deleted_through_the_route(self):
        pet_id = database.create_pet(self.form())
        response = self.client.get(f"/owner/delete/{self.owner_id}")
        self.assertEqual(response.status_code, 400)
        self.assertIn("Cannot delete this owner", response.get_data(as_text=True))
        self.client.get(f"/delete/{pet_id}")
        self.assertEqual(self.client.get(f"/owner/delete/{self.owner_id}").status_code, 302)

    def test_health_reports_foreign_keys(self):
        self.assertEqual(self.client.get("/health").get_data(as_text=True), "ok")


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

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

from peewee import IntegrityError

import database
from database import Pet
import upgrade_database


class ConstraintTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.path = str(Path(self.temp.name) / "pets.db")
        database.initialize(self.path)
        self.owner_id = database.create_owner({"name": "Alex", "city": "Kent"})

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

    def pet(self, **changes):
        values = {"name": "Casey", "type": "dog", "age": "9", "owner_id": self.owner_id}
        values.update(changes)
        return values

    def test_foreign_keys_are_on(self):
        self.assertEqual(database.db.execute_sql("PRAGMA foreign_keys").fetchone()[0], 1)

    def test_missing_owner_is_rejected(self):
        with self.assertRaises(IntegrityError):
            database.create_pet(self.pet(owner_id=999999))
        self.assertEqual(database.get_pets(), [])

    def test_python_checks_come_first(self):
        for values in [self.pet(name=" "), self.pet(type=""), self.pet(age="-1"),
                       self.pet(age="abc"), {"name": "x", "type": "dog"}]:
            with self.assertRaises(ValueError):
                database.create_pet(values)
        self.assertEqual(database.get_pets(), [])

    def test_check_constraints_work_below_the_python_checks(self):
        # Pet.create skips our helper functions, so only the table's rules apply.
        for values in [dict(name="   ", type="dog", age=1),
                       dict(name="Casey", type="  ", age=1),
                       dict(name="Casey", type="dog", age=-1)]:
            with self.assertRaises(IntegrityError):
                Pet.create(owner=self.owner_id, **values)
        self.assertEqual(database.get_pets(), [])

    def test_blank_age_becomes_zero(self):
        pet_id = database.create_pet(self.pet(age=""))
        self.assertEqual(database.get_pet(pet_id)["age"], 0)

    def test_owner_with_pets_cannot_be_deleted(self):
        pet_id = database.create_pet(self.pet())
        with self.assertRaises(IntegrityError):
            database.delete_owner(self.owner_id)
        self.assertIsNotNone(database.get_owner(self.owner_id))
        database.delete_pet(pet_id)
        database.delete_owner(self.owner_id)
        self.assertIsNone(database.get_owner(self.owner_id))

    def test_reassigning_a_pet_frees_the_first_owner(self):
        other = database.create_owner({"name": "Sam"})
        pet_id = database.create_pet(self.pet())
        database.update_pet(pet_id, self.pet(owner_id=other))
        self.assertEqual(database.get_pet(pet_id)["owner_name"], "Sam")
        database.delete_owner(self.owner_id)

    def test_failed_update_leaves_the_pet_unchanged(self):
        pet_id = database.create_pet(self.pet())
        with self.assertRaises(IntegrityError):
            database.update_pet(pet_id, self.pet(name="Renamed", owner_id=999999))
        self.assertEqual(database.get_pet(pet_id)["name"], "Casey")

    def test_atomic_group_rolls_back_after_a_constraint_failure(self):
        with self.assertRaises(IntegrityError):
            with database.db.atomic():
                database.create_pet(self.pet(name="First"))
                database.create_pet(self.pet(name="Second", owner_id=999999))
        self.assertEqual(database.get_pets(), [])

    def test_blank_owner_details_are_stored_as_null(self):
        owner = database.get_owner(database.create_owner({"name": "Sam", "city": " ", "type_of_home": ""}))
        self.assertIsNone(owner["city"])
        self.assertIsNone(owner["type_of_home"])

    def test_pets_list_in_name_then_id_order(self):
        for name in ["Zelda", "Alex", "Alex"]:
            database.create_pet(self.pet(name=name))
        pets = database.get_pets()
        self.assertEqual([p["name"] for p in pets], ["Alex", "Alex", "Zelda"])
        self.assertLess(pets[0]["id"], pets[1]["id"])

    def test_table_definitions_carry_the_rules(self):
        with sqlite3.connect(self.path) as connection:
            sql = connection.execute("select sql from sqlite_master where name='pet'").fetchone()[0]
        for text in ["CHECK (age >= 0)", "CHECK (length(trim(name)) > 0)", "ON DELETE RESTRICT"]:
            self.assertIn(text, sql)


class UpgradeTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.path = str(Path(self.temp.name) / "pets.db")

    def tearDown(self):
        database.close_connection()

    def make(self, script):
        with sqlite3.connect(self.path) as connection:
            connection.executescript(script)

    def rows(self, sql):
        with sqlite3.connect(self.path) as connection:
            return connection.execute(sql).fetchall()

    # Table layouts from the earlier examples.
    TEXT_OWNER = """
        create table pet (id integer primary key autoincrement, name text not null,
            type text not null, age integer, food text, owner text);
        insert into pet values (1, 'Casey', 'dog', 9, 'kibble', 'Alex');
        insert into pet values (2, 'Heidi', 'cat', 15, 'tuna', 'alex');
        insert into pet values (3, 'Suzy', 'mouse', 2, 'seeds', '');
        insert into pet values (4, 'Bad', 'dog', -4, 'x', 'Sam');
    """
    OWNER_TABLE = """
        create table owner (id integer primary key autoincrement, name text not null, city text, type_of_home text);
        create table pet (id integer primary key autoincrement, name text not null, type text not null,
            age integer, food text, owner_id integer not null,
            foreign key (owner_id) references owner(id) on delete restrict);
        insert into owner values (1, 'Alex', 'Kent', 'house');
        insert into owner values (2, 'Sam', null, null);
        insert into pet values (1, 'Casey', 'dog', 9, 'kibble', 1);
        insert into pet values (2, '  ', 'dog', -2, null, 2);
    """
    NO_OWNERS = """
        create table pet (id integer primary key, name text not null, type text not null, age integer, food text);
        insert into pet (name, type, age) values ('Casey', 'dog', 9);
        insert into pet (name, type, age) values ('Felix', 'cat', 3);
    """

    def test_initialize_refuses_an_older_file_and_leaves_it_alone(self):
        self.make(self.NO_OWNERS)
        with self.assertRaises(database.SchemaNotCurrent) as caught:
            database.initialize(self.path)
        self.assertIn("upgrade_database.py", str(caught.exception))
        self.assertEqual(len(self.rows("select * from pet")), 2)
        self.assertEqual(self.rows("select name from sqlite_master where type='table'"), [("pet",)])

    def test_upgrade_of_the_text_owner_layout(self):
        self.make(self.TEXT_OWNER)
        report = upgrade_database.upgrade(self.path)
        self.assertEqual(report["status"], "upgraded")
        self.assertEqual(report["pets"], 3)
        self.assertEqual(report["unassigned"], 1)
        self.assertEqual(report["not_carried"], ["food"])
        self.assertEqual(len(report["skipped"]), 1)
        self.assertIn("age >= 0", report["skipped"][0])
        self.assertEqual(self.rows("select name from owner order by id"), [("Alex",), ("Unassigned",), ("Sam",)])
        self.assertEqual(self.rows("select id, owner_id from pet order by id"), [(1, 1), (2, 1), (3, 2)])
        # The original file is kept, unchanged.
        with sqlite3.connect(report["backup"]) as connection:
            self.assertEqual(connection.execute("select count(*) from pet").fetchone()[0], 4)

    def test_upgrade_of_the_owner_table_layout_keeps_ids_and_skips_bad_rows(self):
        self.make(self.OWNER_TABLE)
        report = upgrade_database.upgrade(self.path)
        self.assertEqual((report["owners"], report["pets"]), (2, 1))
        self.assertEqual(len(report["skipped"]), 1)
        self.assertEqual(self.rows("select pet.id, pet.name, owner.city, pet.owner_id from pet join owner on owner.id = pet.owner_id"),
                         [(1, "Casey", "Kent", 1)])
        self.assertEqual(database.schema_problems(self.path), [])

    def test_upgrade_of_the_layout_without_owners(self):
        self.make(self.NO_OWNERS)
        report = upgrade_database.upgrade(self.path)
        self.assertEqual((report["pets"], report["unassigned"]), (2, 2))
        self.assertEqual(self.rows("select name from owner"), [("Unassigned",)])
        database.initialize(self.path)
        self.assertEqual([p["owner_name"] for p in database.get_pets()], ["Unassigned", "Unassigned"])

    def test_upgraded_tables_enforce_the_rules(self):
        self.make(self.OWNER_TABLE)
        upgrade_database.upgrade(self.path)
        database.initialize(self.path)
        with self.assertRaises(IntegrityError):
            database.db.execute_sql("insert into pet (name, type, age, owner_id) values ('  ', 'dog', -4, 1)")

    def test_current_and_missing_files_are_left_alone_or_created(self):
        self.assertEqual(upgrade_database.upgrade(self.path)["status"], "created")
        self.assertEqual(upgrade_database.upgrade(self.path)["status"], "current")
        self.assertEqual(sorted(p.name for p in Path(self.temp.name).iterdir()), ["pets.db"])

    def test_a_second_upgrade_never_overwrites_the_first_backup(self):
        self.make(self.NO_OWNERS)
        first = upgrade_database.upgrade(self.path)["backup"]
        self.make("drop table pet; create table pet (id integer primary key, name text, type text, age integer)")
        second = upgrade_database.upgrade(self.path)["backup"]
        self.assertNotEqual(first, second)
        self.assertTrue(Path(first).exists() and Path(second).exists())


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

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

"""Build current tables and reload the data from an older pets database.

Usage: python3 upgrade_database.py [database_file]

create_tables() never changes a table that already exists, so a file made by
an earlier example keeps its old rules. This script builds a new file with the
current tables, reloads every row that satisfies the rules, and then swaps the
new file into place. The original file is kept beside it as
<name>.before-upgrade, so nothing is lost.

Files that are already current, and files that do not exist yet, are left
alone or created empty.
"""
import argparse
from datetime import datetime
import os
from pathlib import Path
import sqlite3

from peewee import IntegrityError

import database
from database import Owner, Pet

UNASSIGNED = "Unassigned"
# Old columns this example reads, directly or by turning them into owners.
USED_COLUMNS = {"id", "name", "type", "age", "owner_id", "owner", "city", "type_of_home"}


def read_old_rows(path):
    connection = sqlite3.connect(f"{Path(path).resolve().as_uri()}?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    try:
        tables = {row[0] for row in connection.execute(
            "select name from sqlite_master where type = 'table'")}
        owners = []
        pets = []
        pet_columns = set()
        if "owner" in tables:
            owners = [dict(row) for row in connection.execute("select * from owner order by id")]
        if "pet" in tables:
            pets = [dict(row) for row in connection.execute("select * from pet order by id")]
            pet_columns = {row[1] for row in connection.execute("pragma table_info(pet)")}
        return owners, pets, pet_columns
    finally:
        connection.close()


def clean(value):
    return (value or "").strip() or None


def load_rows(owners, pets, report):
    """Insert what satisfies the rules. Anything else is listed in the report."""
    loaded_owner_ids = set()
    owner_ids_by_name = {}
    for row in owners:
        try:
            with database.db.atomic():
                Owner.create(id=row["id"], name=(row["name"] or "").strip(),
                             city=clean(row.get("city")),
                             type_of_home=clean(row.get("type_of_home")))
        except IntegrityError as error:
            report["skipped"].append(f"owner {row['id']} ({row['name']!r}): {error}")
            continue
        loaded_owner_ids.add(row["id"])
        owner_ids_by_name.setdefault(row["name"].strip().lower(), row["id"])
        report["owners"] += 1

    unassigned_id = None
    for row in pets:
        owner_id = row.get("owner_id")
        if owner_id is None and clean(row.get("owner")):
            # An earlier example kept the owner's name as text on the pet.
            key = row["owner"].strip().lower()
            if key not in owner_ids_by_name:
                owner_ids_by_name[key] = Owner.create(name=row["owner"].strip()).id
                report["owners"] += 1
            owner_id = owner_ids_by_name[key]
        elif owner_id is not None and owner_id not in loaded_owner_ids:
            report["skipped"].append(f"pet {row['id']} ({row['name']!r}): its owner was not loaded")
            continue
        if owner_id is None:
            if unassigned_id is None:
                unassigned_id = Owner.create(name=UNASSIGNED).id
                report["owners"] += 1
            owner_id = unassigned_id
            report["unassigned"] += 1

        try:
            age = 0 if row.get("age") is None else int(row["age"])
            with database.db.atomic():
                Pet.create(id=row["id"], name=(row["name"] or "").strip(),
                           type=(row["type"] or "").strip(), age=age, owner=owner_id)
        except (IntegrityError, ValueError) as error:
            report["skipped"].append(f"pet {row['id']} ({row['name']!r}): {error}")
            continue
        report["pets"] += 1


def backup_name(path):
    backup = Path(f"{path}.before-upgrade")
    if backup.exists():
        backup = Path(f"{path}.before-upgrade-{datetime.now():%Y%m%d%H%M%S}")
    return backup


def upgrade(database_file):
    """Make database_file current. Returns a dictionary describing what happened."""
    path = Path(database_file)
    report = {"status": "current", "owners": 0, "pets": 0, "unassigned": 0,
              "skipped": [], "not_carried": [], "problems": [], "backup": None}

    problems = database.schema_problems(path)
    if not path.exists():
        database.initialize(path)
        database.close_connection()
        report["status"] = "created"
        return report
    if not problems:
        return report

    report["problems"] = problems
    owners, pets, pet_columns = read_old_rows(path)
    report["not_carried"] = sorted(pet_columns - USED_COLUMNS)

    new_path = Path(f"{path}.new")
    new_path.unlink(missing_ok=True)
    try:
        database.initialize(new_path)
        load_rows(owners, pets, report)
    finally:
        database.close_connection()
    backup = backup_name(path)
    os.replace(path, backup)
    os.replace(new_path, path)
    report["backup"] = str(backup)
    report["status"] = "upgraded"
    return report


def print_report(database_file, report):
    if report["status"] == "current":
        print(f"{database_file} already matches the current tables.")
        return
    if report["status"] == "created":
        print(f"{database_file} did not exist. Created empty current tables.")
        return
    print(f"Upgraded {database_file}.")
    for problem in report["problems"]:
        print(f"  found: {problem}")
    print(f"  loaded {report['owners']} owners and {report['pets']} pets")
    if report["unassigned"]:
        print(f"  {report['unassigned']} pets had no owner and now belong to '{UNASSIGNED}'")
    if report["not_carried"]:
        print("  not carried over (no column for them here): " + ", ".join(report["not_carried"]))
    for line in report["skipped"]:
        print(f"  skipped {line}")
    print(f"  original file kept as {report['backup']}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Build current tables and reload an older pets database.")
    parser.add_argument("database_file", nargs="?", default="pets.db")
    args = parser.parse_args()
    print_report(args.database_file, upgrade(args.database_file))