Inleiding¶

Flaskr¶

We werken verder met de Flaskr applicatie uit de vorige lessen. Daar hebben we SQLAlchemy ORM gebruikt om op een declaratieve manier user en post tables te definiëren met een aantal eenvoudige kolommen.

Ter herinneren, voor de user table hebben we het volgende model:

1
2
3
4
5
6
7
8

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

De applicatie werkt perfect maar op een dag beslis jij als ontwikkelaar (of je opdrachtgever) om de applicatie verder uit te breiden.

Voor elke User moet nu een email adres worden bewaard.

Geen probleem, we breiden de User class gewoon uit...

1
2
3
4
5
6
7
8
9

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)
    email: Mapped[str] = mapped_column(nullable=True)

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

Alleen ... hoe zorgen we er voor dat de (elke!) database ook werkelijk deze nieuwe kolom krijgt? Als we Base.metadata.create_all(engine) uitvoeren gebeurt er niets van de tabel bestaat al. Eerste Base.metadata.drop_all(engine) uitvoeren is al zeker geen optie want dan zijn we alle huidige data kwijt.

Precies om dit probleem op te lossen is Alembic gemaakt.

Alembic is een migratie-tool voor SQLAlchemy die helpt om wijzigingen in je database-schema (zoals nieuwe tabellen, kolommen of constraints) op een gecontroleerde en herhaalbare manier door te voeren.

In plaats van manueel SQL-scripts te schrijven, gebruik je Alembic om migraties te genereren, beheren en toepassen via versiebeheer. Zo blijft je database consistent met de evolutie van je Python-modellen in SQLAlchemy.

Voorbereiding¶

Alembic¶

Alembic is een apart Python package en installeer je, bvb. met uv:

uv add alembic

Migration Environment¶

Een migration environment in Alembic is de structuur en configuratie die Alembic nodig heeft om migraties uit te voeren binnen een project.

De basis-structuur van deze migration environment maak je eenmalig aan met het commando alembic init <directory>, waarbij als directory typisch alembic gebruikt wordt.

Je project krijgt dan de bvb. de volgende extra structuur:

yourproject/
    alembic.ini
    pyproject.toml
    alembic/
        env.py
        README
        script.py.mako
        versions/
            3512b954651e_add_account.py
            2b1ae634e5cd_add_order_id.py
            3adcc9a56557_rename_username_field.py

• alembic.ini – Het hoofdconfiguratiebestand.
• alembic/env.py – Het script dat wordt uitgevoerd voor elke migratie.
• alembic/script.py.mako – Een template voor nieuwe migratiebestanden. Mako is een templating library (net als Jinja2!).
• alembic/versions/ – Zal al de gegenereerde migratiebestanden bevatten.

Kort gezegd: het migratie environment is het technische kader waarin Alembic migraties kan aanmaken, bijhouden en toepassen — vergelijkbaar met een “projectconfiguratie” voor databaseversiebeheer.

uv run alembic init alembic

Bekijk de aangemaakte directories en bestanden.

In alembic.ini moet de database URL geconfigureerd worden - op dezelfde manier als deze in SQLAlchemy gebruikt wordt.

1
2

sqlalchemy.url = sqlite:///flaskr/flaskr.sqlite
sqlalchemy.url = postgresql://flaskr@localhost/flaskr

Eerste Migratie: kolom toevoegen¶

We willen dus een age kolom toevoegen aan de user table. De eerste stap is om Alembic een nieuwe revisie de laten aanmaken met het alembic revision commando. Met de -m (message) optie wordt een beschrijving van de revisie meegegeven.

uv run alembic revision -m "Add user.email column"

In de output zien we hoe het eerste migratiebestand wordt gemaakt, bvb. versions/ff3fc27cd408_add_user_email_column.py. De eerste 12 karakters zijn een unieke ID die Alembic zelf genereert.

Het bestand bevat:

• Standaard imports van
  • alembic als op, gebruikt voor handelingen (operations) uit te voeren, zoals het toevoegen van een kolom.
  • sqlalchemy als sa, gebruikt om elementen zoals een nieuwe kolom te definiëren, op dezelfde manier als hoe dit in SQLAlchemy zelf gebeurt.
• Identificatie van de vorige (down_revision) en nieuwe revisie (revision).
• Lege functies upgrade() en downgrade().

Het is dan aan ons om deze functies in te vullen met de code die de gewenste wijzigingen in onze database zullen uitvoeren.

1
2
3
4
5
6
7
8

def upgrade() -> None:
    """Upgrade schema."""
    op.add_column("user", sa.Column("email", sa.String)) # (1)


def downgrade() -> None:
    """Downgrade schema."""
    op.drop_column("user", "email") # (2)

Upgrade¶

Om te upgraden gebruik je het commando alembic upgrade met als argument de versie (revision) naar waar je wil upgraden. Met head upgrade je naar de laatste versie.

uv run alembic upgrade head

We controleren het resultaat:

psql flaskr -c '\d user'

Alembic heeft ook een nieuwe tabel toegevoegd, alembic_version met slechts één kolom en één rij. De waarde is het huidige versie (revisie) en komt overeen met de revisie in ons eerste migratiebestand.

psql flaskr -c 'select * from alembic_version'

Downgrade¶

Om terug te downgraden gebruik je het commando alembic downgrade met ook als argument de versie naar waar je wil downgrade. Met -1 downgrade je naar de vorige revisie. Op dezelfde manier kan +1 gebruikt worden met upgrade om terug naar de volgende revisie te gaan.

uv run alembic downgrade -1

uv run alembic upgrade +1

Tweede Migratie: data migratie¶

Gebruikers van de applicatie willen niet allemaal dat hun gebruikersnaam (username) bij hun posts verschijnt. We willen een nieuwe kolom, displayname, waar gebruikers zelf een alter-ego kunnen instellen. Het model wordt dan:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from datetime import datetime

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)
    email: Mapped[str] = mapped_column(nullable=True)
    displayname: Mapped[str] = mapped_column(nullable=False)

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

Om te starten genereren we een nieuwe Alembic revisie:

uv run alembic revision -m "Add user.displayname column"

Naast het toevoegen van de nieuwe kolom willen we nu ook direct een initiële waarde toevoegen, zijnde username. De kolom moet ook not null zijn.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def upgrade() -> None:
    """Upgrade schema."""
    op.add_column("user", sa.Column("displayname", sa.String)) # (1)
    op.execute(sa.text('UPDATE "user" SET displayname = username')) # (2)
    op.alter_column("user", "displayname", existing_type=sa.String(), nullable=False) # (3)


def downgrade() -> None:
    """Downgrade schema."""
    op.drop_column("user", "displayname")

uv run alembic upgrade head

psql flaskr -c 'select * from "user"'

Andere alembic commando's¶

1
2
3

$ alembic history
ff3fc27cd408 -> 1c0a47692aad (head), Add user.displayname column
<base> -> ff3fc27cd408, Add user.email column

1
2
3
4

$ alembic current
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
1c0a47692aad (head)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

$ alembic current --verbose
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
Current revision(s) for postgresql://flaskr@localhost/flaskr:
Rev: 1c0a47692aad (head)
Parent: ff3fc27cd408
Path: /Users/dfabrice/dev/github/flask-tutorial/alembic/versions/1c0a47692aad_add_user_displayname_column.py

    Add user.displayname column

    Revision ID: 1c0a47692aad
    Revises: ff3fc27cd408
    Create Date: 2025-10-31 08:45:45.282922

Autogenerate¶

Tot nu toe hebben we zelf de link gelegd tussen de aanpassingen in de SQLAlchemy modellen (mapped classes) en de nodige migratiestappen in Alembic.

Alembic kan dit echt ook (deels) automatisch doen met de autogenerate optie.

Hiervoor moet Alembic natuurlijk wel toegang krijgen tot de juiste SQLAlchemy Metadata. Dit doen we in het alembic/env.py script:

1
2
3

from flaskr.models import Base

target_metadata = Base.metadata

Voorbeeld¶

Als voorbeeld voegen we nogmaals een nieuwe kolom toe in User: de created kolom moet bijhouden wanneer een gebruiker zich geregistreerd heeft.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from datetime import datetime

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)
    email: Mapped[str] = mapped_column(nullable=True)
    displayname: Mapped[str] = mapped_column(nullable=False)
    created: Mapped[datetime] = mapped_column(nullable=True)

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

1

alembic revision --autogenerate -m "Add user.created column"

Het resultaat is een migratiebestand met volgende functies:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user', sa.Column('created', sa.DateTime(), nullable=True))
    # ### end Alembic commands ###

def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('user', 'created')
    # ### end Alembic commands ###

Wat kan Autogenerate?¶

Wel¶

• Table en Column toevoegen/verwijderen.
• (Non-)Nullable Column aanpassingen.
• Aanpassingen aan Indexes, Unique Constraints en Foreign Key Constraints.
• Column Type aanpassingen.

Niet¶

• Table name aanpassing. Wordt een drop+create. (!!)
• Column name aanpassing. Idem.
• Anonieme Constraints.
• Speciale SQLAlchemy types, bvb Enum.
• Sequence toevoegen/verwijderen.

Alembic in the praktijk¶

sqlalchemy.url beveiligen¶

Hetalembic.ini bestand gaat in version control en productie-databases zullen (hopelijk) een password nodig hebben.

De sqlalchemy.url kan dynamisch worden ingesteld vanuit het env.py script, bijvoorbeeld met behulp van environment variables.

1
2

if db_url := os.getenv("DB_URL"):
    config.set_main_option("sqlalchemy.url", db_url)

Migratie detecteren¶

Met het alembic check commando kunnen we controleren of een nieuwe migratie revisie nodig is op basis van de huidige aanpassingen in de modellen.

Bvb.

1
2
3
4
5

$ alembic check
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.ddl.postgresql] Detected sequence named 'post_id_seq' as owned by integer column 'post(id)', assuming SERIAL and omitting
No new upgrade operations detected.

1
2
3
4
5
6
7
8

$ alembic check
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.ddl.postgresql] Detected sequence named 'post_id_seq' as owned by integer column 'post(id)', assuming SERIAL and omitting
INFO  [alembic.autogenerate.compare] Detected NOT NULL on column 'user.created'
ERROR [alembic.util.messaging] New upgrade operations detected: [[('modify_nullable', None, 'user', 'created', {'existing_type': TIMESTAMP(), 'existing_server_default': False, 'existing_comment': None}, True, False)]]
  FAILED: New upgrade operations detected: [[('modify_nullable', None, 'user', 'created', {'existing_type': TIMESTAMP(), 'existing_server_default': False, 'existing_comment': None}, True,
  False)]]

Offline mode¶

Met offline mode wordt de migratie niet uitgevoerd maar in de plaats de exacte SQL commando's gegeneerd. Deze kunnen dan bvb. worden gecontroleerd (of zelfs uitgevoerd) door een database administrator (DBA) of specifiek team.

Omdat er geen verbinding wordt gemaakt met de database moeten we zowel de start als de eind revisie meegeven:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

$ alembic history
1c0a47692aad -> 7c0d67786d97 (head), Add user.created column
ff3fc27cd408 -> 1c0a47692aad, Add user.displayname column
<base> -> ff3fc27cd408, Add user.email column

$ alembic upgrade 1c0a47692aad:7c0d67786d97 --sql
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Generating static SQL
INFO  [alembic.runtime.migration] Will assume transactional DDL.
BEGIN;

INFO  [alembic.runtime.migration] Running upgrade 1c0a47692aad -> 7c0d67786d97, Add user.created column
-- Running upgrade 1c0a47692aad -> 7c0d67786d97

ALTER TABLE "user" ADD COLUMN created TIMESTAMP WITHOUT TIME ZONE;

UPDATE alembic_version SET version_num='7c0d67786d97' WHERE alembic_version.version_num = '1c0a47692aad';

COMMIT;

Naamgeving van Constraints¶

Zonder tussenkomst zal de naam van constraints gekozen worden door de database zelf.

Bvb.

1
2
3
4
5
6
7
8

$ psql flaskr -c "select conname, contype from pg_catalog.pg_constraint"
                    conname                     | contype
------------------------------------------------+---------
...
 user_pkey                                      | p
 user_username_key                              | u
 post_pkey                                      | p
 post_author_id_fkey                            | f

Dit is vervelend want:

• Bij aanpassingen (migraties) moeten we de juiste naam gebruiken (bvb. op.drop_constraint()).
• Verschillende databases gebruiken verschillende (automatische) naamgeving.

Oplossing: vaste naming convention in SQLAlchemy/Alembic

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class Base(DeclarativeBase):
    metadata = MetaData(
        naming_convention={
            "ix": "ix_%(column_0_label)s",
            "uq": "uq_%(table_name)s_%(column_0_name)s",
            "ck": "ck_%(table_name)s_%(constraint_name)s",
            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
            "pk": "pk_%(table_name)s",
        }
    )

Zowel SQLAlchemy als Alembic (autogenerate) zullen deze conventies nu gebruiken, ongeacht het type database. We kunnen dit uittesten met bvb. een in-memory SQLite database:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

❯ uv run python3
...
>>> from flaskr.models import Base
>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite:///:memory:", echo=True)
>>> Base.metadata.create_all(engine)
...
CREATE TABLE user (
        id INTEGER NOT NULL,
        username VARCHAR NOT NULL,
        password VARCHAR NOT NULL,
        email VARCHAR,
        displayname VARCHAR NOT NULL,
        CONSTRAINT pk_user PRIMARY KEY (id),
        CONSTRAINT uq_user_username UNIQUE (username)
)

2025-11-01 09:25:13,432 INFO sqlalchemy.engine.Engine [no key 0.00007s] ()
2025-11-01 09:25:13,432 INFO sqlalchemy.engine.Engine
CREATE TABLE post (
        id INTEGER NOT NULL,
        title VARCHAR NOT NULL,
        body TEXT NOT NULL,
        created DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
        author_id INTEGER NOT NULL,
        CONSTRAINT pk_post PRIMARY KEY (id),
        CONSTRAINT fk_post_author_id_user FOREIGN KEY(author_id) REFERENCES user (id)
)

Migratie-script en Hooks¶

Met alembic init krijgen we het standaard migratie-script, env.py, dat gebruikt wordt tijdens de verschillende alembic commando's.

Dit script kan je verder aanpassen binnen je eigen projecten. We hebben dit al gebruikt om de sqlalchemy.url dynamisch in te stellen, en om de juiste SQLAlchemy Metadata mee te geven.

In env.py is context.configure() de centrale plek waar je instelt hoe een migratie uitgevoerd wordt.

Hooks zijn:

• Parameters waarmee Alembic extra logica kan uitvoeren tijdens het migratieproces
• Functies die je meegeeft aan context.configure()

Een paar handiger voorbeelden wat hier mee mogelijk is:

include_object¶

Bvb. om (bepaalde) bestaande tables nooit te droppen.

1

CREATE TABLE foo (bar INT);

1
2
3
4

$ alembic check

  FAILED: New upgrade operations detected: [('remove_table', Table('foo', MetaData(),
  Column('bar', INTEGER(), table=<foo>), schema=None))]

Oplossing:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def include_object(object, name, type_, reflected, compare_to):
    if type_ == "table" and reflected and compare_to is None:
        return False
    else:
        return True

# ...

    context.configure(
        ...
        include_object=include_object,
    )

1
2
3

$ alembic check

No new upgrade operations detected.

process_revision_directives¶

Aangeroepen wanneer je een nieuwe migratie maakt met alembic revision --autogenerate.

Bvb. om lege migratie bestanden te vermijden.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def process_revision_directives(context, revision, directives):
    script = directives[0]
    if script.upgrade_ops.is_empty():
        directives[:] = []

# ...

    context.configure(
        ...
        process_revision_directives=process_revision_directives,
    )

1
2
3

$ alembic revision --autogenerate -m "Test"

# Geen bestand aangemaakt