SQLAlchemy 2 - Toepassen in de flaskr app

Auteur

Fabrice Devaux

Publicatiedatum

16 december 2025

Inleiding

Tijdens deze les passen we de bestaande flaskr applicatie uit Flask 3 - ‘Flaskr’ Tutorial aan om met SQLAlchemy te werken i.p.v. rechtstreeks het Python sqlite3 package te gebruiken.

We gebruiken de ORM met een declaratieve aanpak en volgen grotendeels de werkwijze uit de officiële Flask documentatie.

Flask-SQLAlchemy

Flask-SQLAlchemy is een Flask extensie (package) die voor een iets eenvoudigere configuratie kan zorgen en wat “boilerplate” code op zich neemt.

We gebruiken dit hier niet omdat we de integratie expliciet willen demonstreren, en onze implementatie eenvoudig en puur houden. We willen ook geen extra dependency toevoegen en leren.

Herhaling SQLAlchemy

SQLAlchemy is een ORM (Object Relational Mapper).
- Vertaalt (mapt) automatisch tussen objecten in een programmeeromgeving en tabellen in een relationele databank.
SQLAlchemy Core: Engine, SQL Expression Language, Schema definitie
SQLAlchemy ORM: Core + ORM Mapping

Voorbeeld gebruik met ORM:

from sqlalchemy import (
    ForeignKey,
    Index,
    String,
    UniqueConstraint,
    create_engine,
    select,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[str]

    addresses: Mapped[list["Address"]] = relationship(back_populates="user")

    __table_args__ = (
        UniqueConstraint("name", name="uq_user_name"),
        Index("ix_user_name", "name"),
    )

class Address(Base):
    __tablename__ = "address"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    email_address: Mapped[str]

    user: Mapped[User] = relationship(back_populates="addresses")

engine = create_engine("sqlite:///sqlite.db", echo=True)

Base.metadata.create_all(engine)

walter = User(name="wwhite", fullname="Walter")
walter.addresses.append(Address(email_address="gus.fring@hermanos.com"))

with Session(engine) as session:
    # INSERT
    session.add(walter)
    session.commit()

    # SELECT
    result = session.scalars(select(User).where(User.name == "wwhite")).first()
    print(result.id, result.name)

    # UPDATE
    walter.fullname = "Walter White"
    session.commit()

    # DELETE
    session.delete(walter)
    session.commit()

N+1 Probleem

from sqlalchemy.orm import selectinload

with Session(engine) as session:
    users = session.execute(select(User).options(selectinload(User.addresses))).scalars()
    for user in users:
        print([a.email_address for a in user.addresses])

Installatie

We vertrekken van de werkende implementatie uit de Flask les, waar we rechtstreeks de sqlite3 module hebben gebruikt. Als je twijfelt aan je huidige code gebruik dan https://github.com/dvx76/flask-tutorial als referentie.

Controleer eerst dat je code functioneel is:

cd flask-tutorial
uv run flask --app flaskr run

Het kan interessant zijn de aanpassingen van deze les op een aparte branch te maken zodat je gemakkelijk beide implementaties kan ophalen. Nadat je eventuele laatste aanpassingen gecommit hebt:

git switch main
git switch -c sqlalchemy

Het SQLAlchemy package moet aan het project worden toegevoegd. Bvb. met uv:

uv add "sqlalchemy~=2.0"

Model (Mapped Classes)

Declarative Base en User tabel

Als eerste stap moeten we natuurlijk de database modelleren in SQLAlchemy. Zoals eerder aangegeven kiezen we voor een declaratieve mapping en gebruiken we zoveel mogelijk type hints.

Om de ORM classes te definiëren introduceren we een nieuwe module, models.py, en baseren we ons op het schema zoals het reeds bestaat in schema.sql.

flaskr/models.py

from datetime import datetime
from sqlalchemy import ForeignKey, Text
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(unique=True, nullable=False)
    password: Mapped[str] = mapped_column(nullable=False)

    # posts = relationship("Post", back_populates="author")

Om de starten hebben we altijd een eigen Base subclass van DeclarativeBase nodig.

De user table heeft kolommen met eenvoudige basis types: int en str. Met mapped_column geven we naast het datatype nog een aantal extra eigenschappen mee:

id is the primary key van de table en moet automatisch geïncrementeerd worden.
username moet uniek zijn.
username en password mogen niet null zijn.

Users (en dan bedoelen we User objecten) hebben ook een relatie met hun Posts. Omdat de Post class nog niet bestaat houden we deze lijn nog even gecomment.

Om ze zien welke SQL commandos gegeneerd zullen worden kunnen we volgende eenvoudige test uitvoeren vanuit een Python shell (bvb uv run python) met een in-memory SQLite database:

from flaskr.models import Base
from sqlalchemy import create_engine
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)

We krijgen volgende SQL te zien:

CREATE TABLE user (
        id INTEGER NOT NULL,
        username VARCHAR NOT NULL,
        password VARCHAR NOT NULL,
        PRIMARY KEY (id),
        UNIQUE (username)
)

Dit verschilt enigszins met wat we momenteel in schema.sql hebben maar deze verschillen zijn enkel stilistisch en dus functioneel equivalent.

Oefening

Voeg in models.py de mapped class voor Post toe. Je kan dan ook het posts attribuut in the User class uncommenten.
Test het resultaat op dezelfde manier.

Tips

We hebben reeds al het nodige geïmporteerd.
Om een default waarde te definiëren kan je in mapped_column de server_default parameter gebruiken.
Combineer deze laatste met de sqlalchemy.sql.functions.current_timestamp functie om tot de gewenste SQL code te komen.

Oplossing

flaskr/models.py

class Post(Base):
    __tablename__ = "post"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(nullable=False)
    body: Mapped[str] = mapped_column(Text, nullable=False)
    created: Mapped[datetime] = mapped_column(server_default=current_timestamp())
    author_id: Mapped[int] = mapped_column(ForeignKey("user.id"))

    author = relationship("User", back_populates="posts")

Database module

`init_db`

We willen opnieuw een db.py module die we rechtstreeks kunnen uitvoeren om onze database eenmalig te initialiseren. Omdat we de database integratie helemaal veranderen is het handigste om te starten vanaf nul.

flaskr/db.py

from pathlib import Path
from sqlalchemy import create_engine
from .models import Base

1LOCAL_DIRECTORY = Path(__file__).parent
SQLITE_DB_FILE = str(LOCAL_DIRECTORY / "flaskr.sqlite")
DEFAULT_DATABASE_URL = f"sqlite:///{SQLITE_DB_FILE}"

def init_db(database_url: str = DEFAULT_DATABASE_URL):
2    engine = create_engine(database_url, echo=True)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

3def get_db():
    pass

if __name__ == "__main__":
    init_db()

1: We gebruiken opnieuw pathlib.Path.parent om altijd het absolute pad naar ons SQLite database bestand te krijgen.
2: init_db gebruikt nu SQLAlchemy engine en metadata om het schema aan te maken. Het schema.sql kunnen we nu ook verwijderen.
3: We behouden de get_db functie nog even omdat andere modules deze importeren.

We voeren de initialisatie direct uit. Als je je bestaande SQLite database uit de vorige les wil behouden kan je ook een andere bestandsnaam kiezen (bvb flaskr_sa.sqlite) - of zelfs de initialisatie overslaan.

uv run python -m flaskr.db

SQLAlchemy Engine en Session management

In de vorige implementatie hadden we een sqlite3.Connection voor elke request, bijgehouden in het flask.g object.

Met SQLAlchemy willen we in de plaats voor elke request een sqlalchemy.orm.Session object. Dit kunnen we op twee manieren aanpakken:

Gelijkaardig aan de vorige implementatie een get_session functie die een nieuw Session object aanmaakt en opslaat in bvb. g.db_session. Met in app.teardown_appcontext een functie die g.db_session.close() aanroept.
Één enkel globaal sqlalchemy.orm.scoped_session object gebruiken dat automatisch Sessions voor ons managed. Het is dan wel noodzakelijk via app.teardown_appcontext telkens de remove methode van het scoped_session object aan te roepen.

We kiezen hier voor de tweede optie, enerzijds voor de eenvoud en anderzijds omdat dit het voorbeeld uit de Flask documentatie volgt.

In tegenstelling tot de Flask documentatie willen we echter de database connection string configureerbaar houden d.m.v. de create_app functie. O.a. om gemakkelijk te kunnen testen.

We breiden onze db module uit met een create_db_session functie die we zullen gebruiken in create_app:

flaskr/db.py

from pathlib import Path
from typing import Callable, Optional
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from .models import Base

LOCAL_DIRECTORY = Path(__file__).parent
SQLITE_DB_FILE = str(LOCAL_DIRECTORY / "flaskr.sqlite")
DEFAULT_DATABASE_URL = f"sqlite:///{Path(__file__).parent / 'flaskr.sqlite'}"

def create_db_session(
    database_url: Optional[str] = None,
) -> tuple[scoped_session[Session], Callable]:
1    database_url = database_url if database_url else DEFAULT_DATABASE_URL
    engine = create_engine(database_url)
    db_session = scoped_session(
        sessionmaker(autocommit=False, autoflush=False, bind=engine)
    )

2    def remove_session(_exc=None):
        db_session.remove()

3    return (db_session, remove_session)

1: We houden de database connection string configureerbaar maar gebruiken onze bestaande SQLite URL als default.
2: Een lokaal gedefinieerde functie die gebruikt zal worden in app.teardown_appcontext. Op deze manier kan de functie een referentie naar het scoped_session object meekrijgen.
3: De create_db_session functie geeft zowel het scoped_session object als de remove_session functie terug.

De create_app functie kunnen we dan als volgt aanpassen:

flaskr/__init__.py

from typing import Optional
from flask import Flask
from . import auth, blog, db

def create_app(database_url: Optional[str] = None) -> Flask:
    app = Flask(__name__)
    app.secret_key = "dev"
    app.jinja_options["autoescape"] = True

    db_session, remove_session = db.create_db_session(database_url)
1    app.config["DB_SESSION"] = db_session

2    app.teardown_appcontext(remove_session)
    app.register_blueprint(auth.bp)
    app.register_blueprint(blog.bp)

    return app

1: Het scoped_session object bewaren we in app.config["DB_SESSION"], waar het toegankelijk zal zijn voor alle view functies in the blueprints. Er is dus maar één scoped_session object!
2: De remove_session functie registreren we in app.teardown_appcontext zodat deze op het einde van elke request wordt uitgevoerd.

Op deze manier zijn onze dependencies beter afgelijnd omdat de db module helemaal geen link heeft met Flask zelf (bvb via current_app).

Session opvragen

De oude get_db functie gaf een sqlite3.Connection terug, bewaard in g.db (per request!). Nu hebben we één enkele Session bewaard in de app.config. Om niet telkens de juiste key van die dictionary te moeten onthouden voegen we volgende functie toe in de db module. Als bonus krijgen we dan meteen ook de juiste type hint mee, wat erg handig is als we straks die Session gaan gebruiken. We hernoemen get_db naar get_db_session voor de duidelijkheid.

flaskr/db.py

from flask import current_app

def get_db_session() -> scoped_session[Session]:
    return current_app.config["DB_SESSION"]

Dummy View functies

Door de aanpassingen in __init__.py en db.py werken de view functies (en bijhorende Jinja templates) momenteel niet. Omdat in de functies en templates veelvuldig url_for wordt gebruikt is het handig als we tijdelijk alle view functies vervangen door dummy functies. Op die manier werken alle url_for’s en kunnen we de view functies en templates een voor een aanpakken.

Alle andere code behouden we nog even, als referentie, maar dan wel gecomment.

flaskr/auth.py

@bp.route("/register", methods=("GET", "POST"))
def register():
    return "TODO", 500

@bp.route("/login", methods=("GET", "POST"))
def login():
    return "TODO", 500

@bp.route("/logout")
def logout():
    return "TODO", 500

flaskr/blog.py

@bp.route("/")
def index():
    return "TODO", 500

@bp.route("/create", methods=("GET", "POST"))
def create():
    return "TODO", 500

@bp.route("/<int:id>/update", methods=("GET", "POST"))
def update(id):
    return "TODO", 500

@bp.route("/<int:id>/delete", methods=("POST",))
def delete(id):
    return "TODO", 500

Het zou nu mogelijk moeten zijn een Flask dev server te starten met uv run flask --app flaskr run --debug. Elke route stuurt wel nog een fout terug naar de browser.

De Auth views

Register

Er moet eigenlijk niet zo heel veel aangepast worden. We bekijken hier alle verschillen.

flaskr/auth.py

1import sqlalchemy.exc
2from .db import get_db_session
from .models import User

@bp.route("/register", methods=("GET", "POST"))
def register():
    if request.method == "POST":
        username = request.form["username"]
        password = request.form["password"]
        error = None

        if not username:
            error = "Username is required."
        elif not password:
            error = "Password is required."

        if error is None:
#           db = get_db()
#           try:
#               db.execute(
#                   "INSERT INTO user (username, password) VALUES (?, ?)",
#                   (username, generate_password_hash(password)),
#               )
#               db.commit()
#           except db.IntegrityError:
3            db_session = get_db_session()
4            user = User(username=username, password=generate_password_hash(password))
5            db_session.add(user)
            try:
6                db_session.commit()
            except sqlalchemy.exc.IntegrityError:
                error = f"User {username} is already registered."
            else:
                return redirect(url_for("auth.login"))

        flash(error)

    return render_template("auth/register.html.j2")

1: sqlalchemy.exc hebben we nodig om toegang te krijgen tot exceptions van SQLAlchemy, die een abstractie vormen over de exceptions van elke soort database die SQLAlchemy ondersteunt.
2: De User class zullen we nodig hebben om de te registreren gebruiker (als een object!) te initialiseren. De get_db_session functie voor de SQLAlchemy Session.
3: I.p.v. een sqlite3.Connection via get_db krijgen we nu een SQLAlchemy Session via get_db_session.
4: I.p.v. de form gegevens rechtstreeks in een SQL INSERT statement te plaatsen maken we een User object aan…
5: … en voegen dat vervolgens toe aan de Session.
6: Als de Session gecommit wordt kunnen we steeds een integriteit fout krijgen van de database. Echter zal dit nu een sqlalchemy.exc.IntegrityError i.p.v. een sqlite3.Connection.IntegrityError.

We kunnen de view onmiddellijk testen.

Login

Probeer als oefening zelf de login view uit te werken.

Tips

Gebruik voor de database query Session.scalars i.p.v. Session.execute. Op https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html#orm-queryguide-select-orm-entities lees je meer over de verschillen en voordelen.
Via dezelfde link krijg je direct ter herinnering wat voorbeelden van hoe je een query statement schrijft in SQLAlchemy ORM.
Met de functie one_or_none op het resultaat krijg je, net zoals bij de fetchone functie bij sqlite3, ofwel None, ofwel het eerste resultaat van de query.

Oplossing

flaskr/auth.py

from sqlalchemy import select

@bp.route("/login", methods=("GET", "POST"))
def login():
    if request.method == "POST":
        username = request.form["username"]
        password = request.form["password"]
        error = None
1        db_session = get_db_session()
2        user = db_session.scalars(
            select(User).where(User.username == username)
        ).one_or_none()

        if user is None:
            error = "Incorrect username."
        elif not check_password_hash(user.password, password):
            error = "Incorrect password."

        if error is None:
            session.clear()
3            session["user_id"] = user.id
            return redirect(url_for("blog.index"))

        flash(error)

    return render_template("auth/login.html.j2")

1: Ook hier zullen we een Session nodig hebben.
2: Door scalars i.p.v. execute te gebruiken krijgen we onmiddellijk User objecten terug. De query statement zelf kunnen we nu met sqlalchemy.select en where beschrijven i.p.v. SQL. Met de one_or_none functie op het resultaat krijgen we ofwel het eerste element van het resultaat, ofwel None als geen rows gevonden zijn.
3: Omdat user nu een User object is, en geen tuple of dict met het resultaat van de query, hebben we nu toegang tot attributen zoals id op de object.

Logout, load_logged_in_user en login_required

De logout view en login_required decorator bevatten zelf geen database interactie dus de oorspronkelijke code kan gewoon hergebruikt worden.

De load_logged_in_user functie (automatisch bij elke request uitgevoerd door @bp.before_app_request) kan op dezelfde manier als de login view worden aangepast.

flaskr/auth.py

@bp.before_app_request
def load_logged_in_user():
    user_id = session.get("user_id")

    if user_id is None:
        g.user = None
    else:
        db_session = get_db_session()
        g.user = db_session.scalars(
            select(User).where(User.id == user_id)
        ).one_or_none()

De Blog views

Create

We starten met de create view want alle andere views hebben posts nodig die met deze view gemaakt worden.

flaskr/blog.py

1from .db import get_db_session
from .models import Post

@bp.route("/create", methods=("GET", "POST"))
@login_required
def create():
    if request.method == "POST":
        title = request.form["title"]
        body = request.form["body"]
        error = None

        if not title:
            error = "Title is required."

        if error is not None:
            flash(error)
        else:
2            db_session = get_db_session()
            post = Post(title=title, body=body, author=g.user)
            db_session.add(post)
            db_session.commit()
            return redirect(url_for("blog.index"))

    return render_template("blog/create.html.j2")

1: Net als bij de auth views wordt get_db_session gebruikt om een SQLAlchemy Session te krijgen en zullen we de mapped classes nodig hebben, deze keer voor Post.
2: Het gebruik is gelijkaardig aan de login view maar hier maken we een Post object i.p.v. een User object aan.

In de base en create templates moeten we niets veranderen dus deze create view kunnen we onmiddellijk testen.

Index

De index view is op zichzelf vrij eenvoudig. Voor de Posts query kunnen we op bijna dezelfde manier te werk gaan als tijdens het ophalen van de User in de auth login view. Deze code kan je optioneel als oefening maken.

Om de view te kunnen testen moet je wel eerst de Jinja template aanpassen zoals verder beschreven.

Oplossing

flaskr/blog.py

from sqlalchemy import select

@bp.route("/")
def index():
    db_session = get_db_session()
    posts = db_session.scalars(select(Post))
    return render_template("blog/index.html.j2", posts=posts)

De index template krijgt nu een lijst met Post objecten (eigenlijk een ScalarResult[Post], maar hier kan je over itereren als een lijst) i.p.v. een lijst met database rows (als dictionaries). Bijgevolg zullen we de Jinja code in de template een beetje moeten aanpassen.

flaskr/templates/blog/index.html.j2

{% extends 'base.html.j2' %}

{% block header %}
<h1>{% block title %}Posts{% endblock %}</h1>
{% if g.user %}
<a class="action" href="{{ url_for('blog.create') }}">New</a>
{% endif %}
{% endblock %}

{% block content %}
{% for post in posts %}
<article class="post">
  <header>
    <div>
1      <h1>{{ post.title }}</h1>
      <div class="about">by {{ post.author.username }} on {{ post.created }}</div>
    </div>
    {% if g.user['id'] == post.author_id %}
    <a class="action" href="{{ url_for('blog.update', id=post.id) }}">Edit</a>
    {% endif %}
  </header>
  <p class="body">{{ post.body }}</p>
</article>
{% if not loop.last %}
<hr>
{% endif %}
{% endfor %}
{% endblock %}

1: Omdat elke post in posts nu een volwaardig object is hebben we toegang tot attributen zoals post.title i.p.v. te moeten werken met dictionary keys zoals post['title'].

Opletten

Bekijk aandachtig hoe elk Post object in de template gebruikt wordt. Er is een typisch probleem waar we hier tegen kunnen aanlopen. Tip: bedenkt wat er gebeurt als we eenmaal redelijk wat posts hebben van verschillende users.

Oplossing

We kunnen hier te maken krijgen met het N+1 Probleem!

Met post.author.username krijgen we toegang tot de username via de author relationship van Post. De username komt dus uit een andere table dan de rest van de data. Bijgevolg krijgen we voor elke aparte username een extra database query.

Test dit door een paar verschillende users en bijhorende posts aan te maken en tijdelijk in de db module de echo=True parameter toe te voegen in create_engine.

Een oplossing die we in de SQLAlchemy les bekeken hebben is om bij de query de optie selectinload mee te geven. Een andere goede optie is joinedload. Probeer beide opties uit en bekijk het verschil in de SQLAlchemy echo output.

flaskr/blog.py

from sqlalchemy.orm import joinedload, selectinload

@bp.route("/")
def index():
    db_session = get_db_session()
    posts = db_session.scalars(select(Post).options(selectinload(Post.author)))
    return render_template("blog/index.html.j2", posts=posts)

Het is je misschien opgevallen dat de posts worden weergegeven van oud naar nieuw. We zouden liever de posts in omgekeerde volgorde zien. Probeer dit als oefening op te lossen. Bedenk of dit iets is dat je eerder in Python (of Jinja) code oplost, of in de database query.

Oplossing

In de meeste gevallen is het interessanter of de database dit soort taken (sorteren, filteren, rangschikken, …) rechtstreeks te laten uitvoeren. De database kan dit meestal veel efficiënter.

flaskr/blog.py

@bp.route("/")
def index():
    db_session = get_db_session()
    posts = db_session.scalars(
        select(Post).options(selectinload(Post.author)).order_by(Post.created.desc())
    )
    return render_template("blog/index.html.j2", posts=posts)

Update en de `get_post` functie

flaskr/blog.py

def get_post(id: int) -> Post:
    post = get_db_session().get(Post, id)

    if post is None:
        abort(404, f"Post id {id} doesn't exist.")

    if post.author_id != g.user.id:
        abort(403)

    return post


@bp.route("/<int:id>/update", methods=("GET", "POST"))
@login_required
def update(id):
    post = get_post(id)

    if request.method == "POST":
        title = request.form["title"]
        body = request.form["body"]
        error = None

        if not title:
            error = "Title is required."

        if error is not None:
            flash(error)
        else:
            post.title = title
            post.body = body
            get_db_session().commit()
            return redirect(url_for("blog.index"))

    return render_template("blog/update.html.j2", post=post)

Delete

Probeer ook deze view functie eens als oefening te maken.

Oplossing

flaskr/blog.py

@bp.route("/<int:id>/delete", methods=("POST",))
@login_required
def delete(id):
    get_post(id)
    db_session = get_db_session()
    db_session.delete(get_post(id))
    db_session.commit()
    return redirect(url_for("blog.index"))

Testen

Ook in de unit tests zullen we het gebruik van de database moeten aanpassen naar SQLAlchemy.

Test Setup

Als we de testen toch proberen uitvoeren (uv run pytest tests) krijgen we vooral de volgende fout:

ERROR tests/test_auth.py::test_register_get - sqlalchemy.exc.ArgumentError: Could not parse SQLAlchemy URL from given URL string

Niet onverwacht, als we tests/conftest.py bekijken. Daar moeten we de app fixture aanpassen zodat:

Een correct SQLAlchemy database URL wordt doorgegeven aan create_app en init_db
Data in de database wordt geïnitialiseerd met SQLAlchemy ORM object i.p.v. SQL

tests/conftest.py

1from datetime import datetime
import pytest
from flask import Flask
from flaskr import create_app, db, models

@pytest.fixture
def app():
2    database_url = "sqlite:///file:mem1?mode=memory&cache=shared&uri=True"
    app = create_app(database_url)
    with app.app_context():
        db_session = db.get_db_session()
        db.init_db(database_url)
3        user_test = models.User(
            username="test",
            password="scrypt:32768:8:1$B6EWUB7sblZHpKwE$74951791e0ebcdcf91999e0e4c3e7768fcb87f994c35de0d80e99d83ec36e9f542b76c4d486c57ced5cea72fd76c3f5a64b0a2c31a89a02e3a86a52a6f52fb1c",
        )
        user_other = models.User(
            username="other",
            password="scrypt:32768:8:1$6DDdh5peSL3fmCBy$c997bb4e0ecdeb5ad2e94cd4cde3a58efa290889a7c40497b154cd23df46ffc25f731370f849ab262fd310d9ee48ce5a9950a6910e08b9a8f52ec791acb85c61",
        )
        post = models.Post(
            title="test title",
            body="test\nbody",
            author=user_test,
            created=datetime.fromisoformat("2025-01-01"),
        )
        db_session.add(user_test)
        db_session.add(user_other)
        db_session.add(post)
        db_session.commit()

        yield app

@pytest.fixture
def client(app: Flask):
    return app.test_client()

1: Om een post aan te maken hebben we nu een datetime object nodig (i.p.v. een SQL data string).
2: De juiste database URL moet worden doorgegeven aan create_app en init_db. Met get_db_session krijgen we nu een SQLAlchemy Session i.p.v. een sqlite3.Connection.
3: Data invoeren gebeurt nu met objecten die aan de Session worden toegevoegd (en dan gecommit) - net zoals we in the auth.py en blog.py modules hebben moeten doen.

Auth en Blog testen

De volgende fout die we krijgen is:

FAILED tests/test_auth.py::test_register_post - AttributeError: module 'flaskr.db' has no attribute 'get_db'
FAILED tests/test_blog.py::test_create - AttributeError: module 'flaskr.db' has no attribute 'get_db'

Gelijkaardige aanpassingen zijn nodig in test_auth.py en test_blog.py. We tonen hier enkel de gewijzigde code:

tests/test_auth.py

from sqlalchemy import select
from flaskr import db, models

def test_register_post(client: FlaskClient):
    # ...
    query = select(models.User).where(models.User.username == "a")
    assert db.get_db_session().execute(query).one_or_none() is not None

def test_login(client):
        # ...
        assert g.user.username == "test"

def test_create_update_validate(client: FlaskClient, path: str):
    # ...
    query = select(models.Post).where(models.Post.body == "invalid post")
    assert db.get_db_session().execute(query).one_or_none() is None

def test_delete(client: FlaskClient):
    # ...
    query = select(models.Post).where(models.Post.id == 1)
    assert db.get_db_session().execute(query).one_or_none() is None

tests/test_blog.py

from sqlalchemy import select
from flaskr import db, models

def test_create(client: FlaskClient):
    # ...
    query = select(models.Post).where(models.Post.title == "test_create title")
    post = db.get_db_session().scalars(query).one()
    assert post.body == "test_create body"


def test_update(client: FlaskClient):
    # ...
    query = select(models.Post).where(models.Post.id == 1)
    post = db.get_db_session().scalars(query).one()
    assert post.title == "updated title"
    assert post.body == "updated body"

PostgreSQL

Dankzij SQLAlchemy ORM krijgen we een handige abstractie van de database en kunnen we in onze applicatie met eenvoudige objecten en query syntax werken.

Een bijkomend belangrijk voordeel is dat het nu heel eenvoudig wordt om de eigenlijke database te veranderen, zonder ook maar iets aan onze applicatie-code te moeten aanpassen. In dit laatste hoofdstuk bekijken we hoe we PostgreSQL i.p.v. SQLite kunnen gebruiken. Merk op dat onze unit testen daarbij gewoon blijven werken met een eenvoudige SQLite in-memory database.

Voorbereiding

We hebben uiteraard een PostgreSQL database nodig. Als test gebruiken we een nieuwe user en database, beiden met de naam flaskr

CREATE DATABASE flaskr;
CREATE USER flaskr WITH PASSWORD 'flaskr';
\c flaskr
ALTER DATABASE flaskr OWNER TO flaskr;

Om SQLalchemy met PostgreSQL te gebruiken hebben we ook een zogenaamde driver (of adapter) nodig. M.a.w. een package die de interface naar een PostgreSQL database implementeert. Er zijn verschillende opties maar psycopg2 is de populairste dus deze voegen we toe aan ons project:

uv add psycopg2

Database URL

Het enige wat we nu moeten doen de database URL die SQLAlchemy gebruikt aanpassen. Als eerste test passen we de DEFAULT_DATABASE_URL in db.py als volgt aan:

flaskr/db.py

DEFAULT_DATABASE_URL = "postgresql://flaskr:flaskr@localhost:5432/flaskr"

Het database schema wordt eenmalig aangemaakt en dan kan Flask gestart worden.

uv run python -m flaskr.db

uv run flask --app flaskr run --debug

Configuratie verbeteren

De database URL hardcoden is natuurlijk geen goed idee. Enerzijds veiligheidsredenen, maar we willen misschien naast de productie-omgeving ook een staging-omgeving.

Er zijn heel wat manieren waarmee we dit kunnen aanpakken. We kiezen hier voor een vrij eenvoudige aanpak die in lijn is met de voorbeelden en best practices uit de Flask documentatie.

Als eerste stap centraliseren we de default configuratie in een nieuwe module. De default configuratie gebruikt terug SQLite. Op die manier kunnen ontwikkelaars gemakkelijk lokaal testen.

flaskr/settings.py

from pathlib import Path

DATABASE_URL = f"sqlite:///{Path(__file__).parent / 'flaskr.sqlite'}"
SECRET_KEY = "dev"

Deze kunnen we door de Flask app laten inlezen als volgt:

flaskr/__init__.py

from flask import Flask
from . import auth, blog, db

def create_app() -> Flask:
    app = Flask(__name__)
    app.config.from_object("flaskr.settings")
    app.jinja_options["autoescape"] = True

    db_session, remove_session = db.create_db_session(app.config["DATABASE_URL"])
    app.config["DB_SESSION"] = db_session

    app.teardown_appcontext(remove_session)
    app.register_blueprint(auth.bp)
    app.register_blueprint(blog.bp)

    return app

Als we na app.config.from_object("flaskr.settings") ook app.config.from_envvar("FLASKR_SETTINGS") toevoegen dan wordt ook de configuratie uit het bestand gedefinieerd in de FLASKR_SETTINGS environment variable gebruikt.

Bijvoorbeeld:

flaskr.settings

DATABASE_URL = "postgresql://flaskr:flaskr@localhost:5432/flaskr"
SECRET_KEY = "192b9bdd22ab9ed4d12e236c78afcb9a393ec15f71bbf5dc987d54727823bcbf"

(voeg *.settings toe aan .gitignore!)

MacOS/Linux
Windows CMD

export FLASKR_SETTINGS=$(pwd)/flaskr.settings
uv run flask --app flaskr run --debug

set FLASKR_SETTINGS=%cd%\flaskr.settings
uv run flask --app flaskr run --debug

Test initialisatie

Als laatste stap moeten we ook de test initialisatie in conftest.py aanpakken.

I.p.v. enkel de database_url laten we create_app een volledige configuratie (als dictionary) meekrijgen, die met app.config.from_mapping wordt ingeladen. Dit gebeurt bovenop alle vorige configuratie en op die manier kunnen we zeker zijn dat tijdens de testen de test configuratie gebruikt wordt.

flaskr/__init__.py

from typing import Any, Mapping, Optional
from flask import Flask
from . import auth, blog, db

def create_app(test_config: Optional[Mapping[str, Any]] = None) -> Flask:
    app = Flask(__name__)
    app.config.from_object("flaskr.settings")
    app.config.from_envvar("FLASKR_SETTINGS", silent=True)
    if test_config:
        app.config.from_mapping(test_config)
    app.jinja_options["autoescape"] = True

    db_session, remove_session = db.create_db_session(app.config["DATABASE_URL"])
    app.config["DB_SESSION"] = db_session

    app.teardown_appcontext(remove_session)
    app.register_blueprint(auth.bp)
    app.register_blueprint(blog.bp)

    return app

tests/conftest.py

@pytest.fixture
def app():
    database_url = "sqlite:///file:mem1?mode=memory&cache=shared&uri=True"
    app = create_app({"DATABASE_URL": database_url, "SECRET_KEY": "test"})
    # ...

Extras

Referentie Implementatie

Een volledig werkende implementatie gebaseerd op deze cursus vind je op https://github.com/dvx76/flask-tutorial/tree/sqlalchemy.

Inleiding

Herhaling SQLAlchemy

Installatie

Model (Mapped Classes)

Declarative Base en User tabel

Oefening

Database module

init_db

SQLAlchemy Engine en Session management

Session opvragen

Dummy View functies

De Auth views

Register

Login

Logout, load_logged_in_user en login_required

De Blog views

Create

Index

Update en de get_post functie

Delete

Testen

Test Setup

Auth en Blog testen

PostgreSQL

Voorbereiding

Database URL

Configuratie verbeteren

Test initialisatie

Extras

Referentie Implementatie

`init_db`

Update en de `get_post` functie