Database Abstraction: Code

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

import database

app = Flask(__name__)
database.setup_database("pets.db")


@app.route("/", methods=["GET"])
@app.route("/hello", methods=["GET"])
@app.route("/hello/<name>", methods=["GET"])
def get_hello(name="world"):
    return render_template("hello.html", name=name)


@app.route("/bye", methods=["GET"])
def get_bye():
    return "Bye!"


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


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


@app.route("/create", methods=["POST"])
def post_create():
    database.create_pet(dict(request.form))
    return redirect(url_for("get_pets"))


@app.route("/update", methods=["GET"])
@app.route("/update/<int:id>", methods=["GET"])
def get_update(id=None):
    if id is None:
        return render_template("error.html", error_message="No ID was provided."), 400
    data = database.get_pet(id)
    if data is None:
        return render_template("error.html", error_message="Data not found."), 404
    return render_template("update.html", data=data)


@app.route("/update", methods=["POST"])
@app.route("/update/<int:id>", methods=["POST"])
def post_update(id=None):
    if id is None:
        return render_template("error.html", error_message="No ID was provided."), 400
    if database.get_pet(id) is None:
        return render_template("error.html", error_message="Data not found."), 404
    database.update_pet(id, dict(request.form))
    return redirect(url_for("get_pets"))


@app.route("/delete/<int:id>", methods=["GET"])
def get_delete(id):
    database.delete_pet(id)
    return redirect(url_for("get_pets"))

import sqlite3
from pprint import pprint

connection = None

def initialize(database_file):
    global connection
    connection = sqlite3.connect(database_file, check_same_thread=False)
    connection.row_factory = sqlite3.Row
    print("succeeded in making connection.")

def setup_database(database_file):
    initialize(database_file)
    connection.execute(
        """create table if not exists pet (
            id integer primary key autoincrement,
            name text not null,
            type text not null,
            age integer,
            food text,
            owner text
        )"""
    )
    columns = {row["name"] for row in connection.execute("pragma table_info(pet)")}
    for column in ("food", "owner"):
        if column not in columns:
            connection.execute("alter table pet add column " + column + " text")
    connection.commit()


def get_pets():
    cursor = connection.cursor()
    cursor.execute("""select * from pet order by id""")
    pets = cursor.fetchall()
    pets = [dict(pet) for pet in pets]
    for pet in pets:
        print(pet)
    return pets

def get_pet(id):
    id = int(id)
    cursor = connection.cursor()
    cursor.execute("""select * from pet where id = ?""", (id,))
    pet = cursor.fetchone()
    if pet is None:
        return None
    return dict(pet)


def create_pet(data):
    try:
        data["age"] = int(data["age"])
    except:
        data["age"] = 0
    cursor = connection.cursor()
    cursor.execute(
        """insert into pet(name, age, type, food, owner) values (?,?,?,?,?)""",
        (data["name"], data["age"], data["type"], data["food"], data["owner"]),
    )
    connection.commit()

def delete_pet(id):
    id = int(id)
    cursor = connection.cursor()
    cursor.execute("""delete from pet where id = ?""", (id,))
    connection.commit()

def update_pet(id, data):
    try:
        data["age"] = int(data["age"])
    except:
        data["age"] = 0
    cursor = connection.cursor()
    cursor.execute(
        """update pet set name=?, age=?, type=?, food=?, owner=? where id=?""",
        (data["name"], data["age"], data["type"], data["food"], data["owner"], id),
    )
    connection.commit()

def setup_test_database():
    setup_database("test_pets.db")
    connection.execute("delete from pet")
    connection.commit()
    pets = [
        {"name": "dorothy", "type": "dog", "age": 9, "food": "pet food", "owner": "greg"},
        {"name": "suzy", "type": "mouse", "age": 9, "food": "pet food", "owner": "greg"},
        {"name": "casey", "type": "dog", "age": 9, "food": "pet food", "owner": "greg"},
        {"name": "heidi", "type": "cat", "age": 15, "food": "tuna", "owner": "david"},
    ]
    for pet in pets:
        create_pet(pet)
    pets = get_pets()
    assert len(pets) == 4


def test_get_pets():
    print("testing get_pets()")
    pets = get_pets()
    assert type(pets) is list
    assert type(pets[0]) is dict
    for key in ["name", "age", "type", "food", "owner"]:
        assert key in pets[0]
        assert type(pets[0]["name"]) == str 


def test_create_pet():
    print("testing create_pet()")
    create_pet({"name": "Food test", "age": 2, "type": "cat",
                "food": "salmon", "owner": "alex"})
    pet = get_pets()[-1]
    assert pet["food"] == "salmon"
    assert pet["owner"] == "alex"
    pet["food"] = "tuna"
    pet["owner"] = "sam"
    update_pet(pet["id"], pet)
    saved = get_pet(pet["id"])
    assert saved["food"] == "tuna"
    assert saved["owner"] == "sam"
    delete_pet(pet["id"])
    assert get_pet(pet["id"]) is None


if __name__ == "__main__":
    setup_test_database()
    test_get_pets()
    test_create_pet()
    print("done.")

# initialize("pets.db")
# cursor = connection.execute("select * from pet")
# rows = [dict(row) for row in cursor.fetchall()]
# pprint(rows)

# These is a tuple
# ("a",1,"b",2)
# This is a dictionary
# { "name": "a","age":1,"kind":"b","whatever":2 }

"""Pet database operations using one JSON file per record.

Change app.py to: import json_database_layer as database
The existing setup_database("pets.db") call selects pets_json, leaving SQLite alone.
An explicit folder is also accepted: setup_database("practice_records").
"""
from pathlib import Path
from jsonbase import JsonBase

store = None


def initialize(database_file):
    global store
    folder = Path(database_file)
    if folder.suffix == ".db":
        folder = folder.with_name(folder.stem + "_json")
    store = JsonBase(folder)


def setup_database(database_file):
    initialize(database_file)


def get_pets():
    return store.get_all()


def get_pet(id):
    return store.get(id)


def pet_record(data):
    try:
        age = int(data["age"])
    except (KeyError, TypeError, ValueError):
        age = 0
    return {"name": data["name"], "age": age, "type": data["type"],
            "food": data["food"], "owner": data["owner"]}


def create_pet(data):
    store.create(pet_record(data))


def update_pet(id, data):
    store.update(id, pet_record(data))


def delete_pet(id):
    store.delete(id)

"""Store dictionaries as numbered JSON files in a specified folder."""
import json
from pathlib import Path
import tempfile


class JsonBase:
    def __init__(self, folder):
        self.folder = Path(folder)
        self.folder.mkdir(parents=True, exist_ok=True)
        self.counter = self.folder / ".next-id"

    def _path(self, id):
        return self.folder / f"{int(id)}.json"

    def _write(self, path, text):
        # Write completely before replacing the destination.
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8",
                                         dir=self.folder, delete=False) as output:
            temporary = Path(output.name)
            try:
                output.write(text)
            except BaseException:
                temporary.unlink(missing_ok=True)
                raise
        try:
            temporary.replace(path)
        finally:
            temporary.unlink(missing_ok=True)

    def get_all(self):
        paths = sorted(self.folder.glob("[0-9]*.json"), key=lambda p: int(p.stem))
        return [self.get(path.stem) for path in paths]

    def get(self, id):
        path = self._path(id)
        if not path.exists():
            return None
        return json.loads(path.read_text(encoding="utf-8"))

    def create(self, data):
        existing = [int(p.stem) for p in self.folder.glob("[0-9]*.json")]
        next_id = int(self.counter.read_text()) if self.counter.exists() else 1
        id = max(next_id, max(existing, default=0) + 1)
        text = json.dumps(dict(data, id=id), indent=2, ensure_ascii=False) + "\n"
        self._write(self.counter, str(id + 1))
        self._write(self._path(id), text)
        return id

    def update(self, id, data):
        path = self._path(id)
        if path.exists():
            text = json.dumps(dict(data, id=int(id)), indent=2, ensure_ascii=False) + "\n"
            self._write(path, text)

    def delete(self, id):
        self._path(id).unlink(missing_ok=True)

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

import argparse
import database

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Create or update the pet database.")
    parser.add_argument("database_file", nargs="?", default="pets.db")
    args = parser.parse_args()
    database.setup_database(args.database_file)
    database.connection.close()
    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>
            <p>Owner:<input name="owner"/></p>
            <hr/>
            <button type="submit">Create</button>
            <a href="/list">Cancel</a>
            <hr/>
        </form>
    </body>
</html>

<html>
    <head>
        <title>Pet List</title>
        <meta charset="utf-8">
        <meta name="viewport" content="width=device-width, initial-scale=1">
        <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
        <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
    </head>
    <body><H1>ERROR</H1>
        <hr/>
        <em>{{error_message}}</em>
        <hr>
        <a href="/list">OK</a>
    </body>
</html>

<html>
    <body>Hello, {{ name }}! (from the template)</body>
</html>

<html>
    <h3>List:</h3>
    <table>
        <tr>
            <th>ID</th>
            <th>Name</th>
            <th>Type</th>
            <th>Age</th>
            <th>Food</th>
            <th>Owner</th>
        </tr>
        {% for pet in pets %}
        <tr>
            {% for key in ['id', 'name', 'type', 'age', 'food', 'owner'] %}
            <td>{{ pet[key] if pet[key] is not none else '' }}</td>
            {% endfor %}
            <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>
            <p>Owner:<input name="owner" value="{{data['owner'] or ''}}"/></p>
            <hr/>
            <button type="submit">Update</button>
            <a href="/list">Cancel</a>
            <hr/>
        </form>
    </body>
</html>

"""Run with python3 -m unittest -v test_json_database.py."""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from jsonbase import JsonBase
import json_database_layer as database


class JsonDatabaseTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.folder = Path(self.temp.name) / "records"

    def test_records_and_reopening(self):
        store = JsonBase(self.folder)
        id = store.create({"name": "O'Malley", "food": "tuna"})
        self.assertEqual(json.loads((self.folder / f"{id}.json").read_text())["food"], "tuna")
        reopened = JsonBase(self.folder)
        self.assertEqual(reopened.get(str(id))["name"], "O'Malley")
        reopened.update(id, {"id": 900, "name": "Changed"})
        self.assertEqual(store.get(id), {"id": id, "name": "Changed"})
        reopened.delete(id)
        self.assertIsNone(store.get(id))
        self.assertEqual(store.get_all(), [])
        self.assertGreater(JsonBase(self.folder).create({"name": "Next"}), id)

    def test_separate_folders_and_missing_records(self):
        first = JsonBase(self.folder)
        second = JsonBase(Path(self.temp.name) / "other")
        first.create({"name": "A"})
        second.update(99, {"name": "Absent"})
        second.delete(99)
        self.assertIsNone(second.get(99))
        self.assertEqual(second.get_all(), [])

    def test_failed_serialization_preserves_record(self):
        store = JsonBase(self.folder)
        id = store.create({"name": "Original"})
        with self.assertRaises(TypeError):
            store.update(id, {"name": object()})
        self.assertEqual(store.get(id)["name"], "Original")

    def test_adapter_and_sqlite_file_preserved(self):
        path = Path(self.temp.name) / "pets.db"
        path.write_bytes(b"leave this file alone")
        database.setup_database(path)
        data = dict(name="Dorothy", type="dog", age="bad", food="kibble", owner="Greg")
        self.assertIsNone(database.create_pet(data))
        pet = database.get_pets()[0]
        self.assertEqual(pet["age"], 0)
        data.update(age="12", food="salmon", owner="Sam")
        self.assertIsNone(database.update_pet(pet["id"], data))
        database.setup_database(path)
        saved = database.get_pet(pet["id"])
        self.assertEqual((saved["age"], saved["food"], saved["owner"]), (12, "salmon", "Sam"))
        self.assertIsNone(database.delete_pet(pet["id"]))
        self.assertIsNone(database.get_pet(pet["id"]))
        self.assertEqual(path.read_bytes(), b"leave this file alone")

    def test_sqlite_order_is_explicit(self):
        source = Path(__file__).resolve().parent
        program = """import database
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as folder:
    database.setup_database(str(Path(folder) / 'pets.db'))
    for name in ('First', 'Second'):
        database.create_pet(dict(name=name, type='cat', age=1, food='kibble', owner='Sam'))
    database.connection.execute('PRAGMA reverse_unordered_selects = ON')
    pets = database.get_pets()
    assert [pet['id'] for pet in pets] == sorted(pet['id'] for pet in pets)
    database.test_create_pet()
    database.connection.close()
"""
        result = subprocess.run([sys.executable, "-B", "-c", program], cwd=source,
                                capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_same_app_with_both_backends(self):
        source = Path(__file__).resolve().parent
        for backend in ("database", "json_database_layer"):
            with self.subTest(backend=backend):
                work = Path(self.temp.name) / backend
                work.mkdir()
                for name in ("app.py", "database.py", "jsonbase.py", "json_database_layer.py"):
                    shutil.copy2(source / name, work / name)
                shutil.copytree(source / "templates", work / "templates")
                p = work / "app.py"
                p.write_text(p.read_text().replace("import database\n", f"import {backend} as database\n", 1))
                script = '''import app
c = app.app.test_client()
for path in ('/', '/pets', '/list', '/create'):
    assert c.get(path).status_code == 200
assert c.get('/update').status_code == 400
assert c.get('/update/999').status_code == 404
assert c.post('/update/999', data={}).status_code == 404
app.database.create_pet(dict(name='Missing fields', type='cat', age=4,
                             food=None, owner=None))
missing = app.database.get_pets()[0]
form = c.get('/update/'+str(missing['id'])).data
assert b'name="food" value=""' in form
assert b'name="owner" value=""' in form
assert b'value="None"' not in form
app.database.delete_pet(missing['id'])
data = dict(name="O'Malley", type='cat', age='4', food='tuna', owner='Alex')
assert c.post('/create', data=data).status_code == 302
pet = app.database.get_pets()[0]
assert c.get('/update/'+str(pet['id'])).status_code == 200
assert b'O'Malley' in c.get('/list').data
data.update(food='salmon', owner='Sam', age='5')
assert c.post('/update/'+str(pet['id']), data=data).status_code == 302
saved = app.database.get_pet(pet['id'])
assert (saved['food'], saved['owner'], saved['age']) == ('salmon', 'Sam', 5)
'''
                restart = '''import app
pet = app.database.get_pets()[0]
assert (pet['food'], pet['owner'], pet['age']) == ('salmon', 'Sam', 5)
c = app.app.test_client()
assert c.get('/delete/'+str(pet['id'])).status_code == 302
assert app.database.get_pet(pet['id']) is None
assert app.database.get_pets() == []
'''
                for program in (script, restart):
                    result = subprocess.run([sys.executable, "-B", "-c", program], cwd=work,
                                            capture_output=True, text=True)
                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)


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

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