SQLAlchemy

Für ein kommendes Projekt möchte ich eine größere Menge Daten speichern, die miteinander in verschiedenen Beziehungen stehen oder voneinander abhängig sind. Daten an sich kann man in einer Text- oder Binärdatei ohne große Probleme speichern (z.B. mittels pickle). Doch wenn bestimmte Datensätze gesucht werden sollen empfiehlt es sich auf eine Datenbank zurückzugreifen.

Hierfür gibt es bereits Module in Python integriert sind (z.B. sqlite3). Doch mangelt es diesen Modulen an der Benutzerfreundlichkeit, da man jede Datenabfrage selbst als SQL-Abfrage formulieren und auswerten muss.

Das Modul SQLAlchemy bietet hingegen eine benutzerfreundliche Schnittstelle zwischen Datenstrukturen und einer Datenbank. Ziel ist es somit, anstelle von SQL-Abfragen, einfach ein Objekt bzw. eine Klasseninstanz zu manipulieren und diese dann der Datenbank zu übergeben.

Im nachfolgenden fasse ich meinen ersten Kontakt und das, was ich über SQLAlchemy gelernt habe zusammen.

Grundprinzipien

SQLAlchemy kann auf zwei Weisen verwendet werden. Entweder man nutzt SQLAlchemy ORM (Object-Relational Mapper), wobei der ORM Muster für die Datenzuordnung zwischen Datenbank und Klassen bereitstellt, oder man benutzt SQLAlchemy Core, was der klassischen Handhabung von Datenbanken näher kommt. Während mit dem ORM Einträge einer Tabelle als Instanzen einer Klasse dargestellt werden, muss man unter Core die Übersetzung der Datenbankstruktur zur Objekten unter Python selber vornehmen.

In beiden Fällen werden Beziehungen zwischen den Tabellen mit Primärschlüsseln (primary key) und Fremdschlüsseln (foreign key) gekennzeichnet und gespeichert. Aber will man auf das Bezugsobjekt zugreifen, muss man es unter ORM eben nicht extra aus der Datenbank auslesen, sondern kann direkt über das von SQLAlchemy zurückgegebene Objekt darauf zugreifen. Dies wird in dem nachfolgenden Beispiel deutlicher, wo eine überschaubare Datenbank zur Verwaltung von Einträgen eines Todo Programms erstellt und mit Beispieldaten gefüllt wird. Am Ende dieses Artikels erläutere ich auch kurz, wie eine vergleichbare Datenbankstruktur unter Core erstellt.

Setup

Zunächst muss SQLAlchemy installiert werden. Dies kann mittels PIP wie folgt getan werden:

pip install SQLAlchemy

Im Code muss dann zuerst eine Verbindung zu einer Datenbank erstellt werden. Der Datenbanktyp ist beliebig, aber für Beispielzwecke eignet sich eine lokale Datenbank:

from sqlalchemy import create_engine

engine = create_engine("sqlite:///database.db")

Oder man erstellt eine Datenbank im Zwischenspeicher, wodurch keine Datei erstellt wird - die Daten aber nicht dauerhaft gespeichert werden:

from sqlalchemy import create_engine

engine = create_engine("sqlite:///:memory:")

Diese Schritte sind sowohl für ORM als auch für Core dieselben.

ORM

Man kann sich merken, dass alle in Python definierte Klassen deklaratorisch sind und somit automatisch das Schema einer Tabelle definieren. Hierbei geben die Klassenparameter und ihre Datentypen jeweils den Spaltennamen und Datentypen (entsprechend in SQLAlchemy) an. Um die Klassen als deklaratorisch zu definieren, basieren sie auf der Klasse DeclarativeBase, und um die Datentypen anzugeben, werden Typenhinweise gemäß PEP 484 eingesetzt:

from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
)
from typing import Optional
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users_table"

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

    def __repr__(self) -> str:
        return f"<User(id={self.id}, name={self.name})>"

class Todo(Base):
    __tablename__ = "todos_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    description: Mapped[str] = mapped_column(nullable=False)
    done_date: Mapped[Optional[datetime]] = mapped_column(nullable=True)

    def __repr__(self) -> str:
        return f"<Todo(id={self.id}, desc={self.description}, done_date={self.done_date})>"

Im vorstehenden Code werden also zwei Klassen definiert, die in Tabellen übersetzt werden. Der Tabellenname wird mit __tablename__ festgelegt und mit dem gemappten Typenhinweis (mit Mapped[_T]) bzw. der Funktion mapped_column() werden die Eigenschaften der Tabellenspalten beschrieben. Dass eine Spalte einen Primärschlüssel (mit primary_key=True) sein, nicht wiederholende Werte (mit unique=True) beinhalten, oder nicht leere Einträge (mit nullable=False) enthalten soll, wird hier recht anschaulich gezeigt. Man kann auch bei den Typenhinweisen angeben, dass Einträge nicht zwingend sind (mit Optional[_T]); z.B. wenn das Todo-Element noch nicht abgeschlossen ist, hat es noch kein Abschlussdatum, was durch den Wert Null gekennzeichnet wird.

Mag man auf die Typenhinweise nicht zurückgreifen, kann man die Datentypen auch in der Funktion mapped_column() als erstes Argument definieren.

Die Methode __repr__(self) erleichtern die Lesbarkeit einer Instanz der Klasse, wenn sie z.B. mit print() in der Befehlszeile ausgegeben wird.

Um nun eine Datenbank mit ihren Tabellen zu erzeugen, muss die Methode create_all() der Klasse Base aufgerufen werden, und man kann die ersten Daten einfügen:

from sqlalchemy.orm import sessionmaker
from datetime import datetime

def main() -> None:
    Base.metadata.create_all(engine)

    Session = sessionmaker(bind=engine)

    with Session() as session:
        user_max = User(name="Max")
        user_kathi = User(name="Kathi")
        session.add(user_max)
        session.add(user_kathi)

        todo_1 = Todo(description="Write code", done_date=datetime.now())
        todo_2 = Todo(description="Tidy your room")
        session.add(todo_1)
        session.add(todo_2)

        session.commit()

        print(session.query(User).all())
        print(session.query(Todo).all())

if __name__ == "__main__":
    main()

Führt man den vorstehenden Code nun aus, erhält man das folgende Ergebnis:

[<User(id=1, name=Max)>, <User(id=2, name=Kathi)>]
[<Todo(id=1, desc=Write code, done_date=2024-10-12 17:55:10.716922)>, <Todo(id=2, desc=Tidy your room, done_date=None)>]

Wie man sieht, können Daten recht einfach erzeugt und gespeichert werden. Hierbei muss immer eine Session erzeugt werden, die durch den Kontextmanager with auch geschlossen wird (sonst müsste man dies mit session.close() selber machen, was gerne vergessen wird). Mit session.add() werden die neuen Datensätze automatisch in die richtige Tabelle eingefügt, und mit session.commit() werden die Änderungen in die Datenbank geschrieben. Die Abfrage der Daten geschieht dann mit session.query(), wobei in diesem Beispiel einfach alle Daten angezeigt werden sollen (mit .all()). Möchte man einen Eintrag ändern, muss man nur den Parameter des entsprechenden Objekts ändern, die Änderung mit session.commit() anwenden und das Objekt mit session.refresh() aktualisieren. Mag man z.B. den Namen eines Benutzers ändern, tut man dies wie folgt:

...
    user_max.name = "Maximilian"
    session.commit()
    session.refresh(user_max)
...

Und wenn man den Benutzer erst aus der Datenbank auslesen muss, kann man dies mit einer Abfrage machen, die mit dem Zusatz .where() oder .filter() die zurückzugebenden Werte beschränkt:

...
    user_max = session.query(User).where(User.id == 1).first()
    # or
    user_max = session.query(User).where(User.name == "Max").first()
...

Der Eintrag des Benutzers hätte nun von <User(id=1, name=Max)> zu <User(id=1, name=Maximilian)> geändert.

Das Löschen eines Eintrags geschieht mit session.delete(), wobei der zu löschende Eintrag als Objekt übergeben wird und die Löschung mit session.commit() angewendet wird.

Diese Daten stehen aber noch in keiner Beziehung zueinander. Dies ändert man, indem die Fremdschlüssel (mit dem Argument ForeignKey()) und Beziehungen (mit der Funktion relationship()) zu den Klassenparametern hinzugefügt werden. Der letztendliche Code sieht dann wie folgt aus:

from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import (
    sessionmaker,
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
)
from typing import List, Optional
from datetime import datetime

engine = create_engine("sqlite:///:memory:")

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False, unique=True)
    todos: Mapped[List["Todo"]] = relationship(back_populates="user")

    def __repr__(self) -> str:
        return f"<User(id={self.id}, name={self.name}, todos={len(self.open_todos)}/{len(self.todos)})>"

    @property
    def open_todos(self) -> List["Todo"]:
        return [t for t in self.todos if t.done_date is None]

class Todo(Base):
    __tablename__ = "todos_table"

    id: Mapped[int] = mapped_column(primary_key=True)
    description: Mapped[str] = mapped_column(nullable=False)
    done_date: Mapped[Optional[datetime]] = mapped_column(nullable=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users_table.id"), nullable=False)
    user: Mapped["User"] = relationship(back_populates="todos")

    def __repr__(self) -> str:
        return f"<Todo(id={self.id}, desc={self.description}, user={self.user.name}, done_date={self.done_date})>"

def main() -> None:
    Base.metadata.create_all(engine)

    Session = sessionmaker(bind=engine)

    with Session() as session:
        user_max = User(name="Max")
        user_kathi = User(name="Kathi")
        session.add(user_max)
        session.add(user_kathi)

        todo_1 = Todo(description="Write code", user=user_max, done_date=datetime.now())
        todo_2 = Todo(description="Tidy your room", user=user_max)
        session.add(todo_1)
        session.add(todo_2)

        session.commit()

        print(session.query(User).all())
        print(session.query(Todo).all())

if __name__ == "__main__":
    main()

In diesem Fall ist die Klasse User um den Parameter todos erweitert worden, die eine Liste von Instanzen der Klasse Todo enthalten soll (mit Mapped[List["Todo"]]). Da diese Liste keine Werte an sich enthält, sondern nur eine vereinfachte Zugriffsmöglichkeit auf Todo-Elemente bereitstellen soll, wird hier auch keine Spalte definiert, sondern eine Beziehung (mit relationship(back_populates="user")). Dies wirkt vielleicht etwas kryptisch auf den ersten Blick. Allerdings bedeutet das Argument der Funktion nur, dass der Parameter (bzw. die Spalte) user der Klasse Todo zurück auf die Klasse User verweisen soll.

Dementsprechend ist in der Klasse Todo auch der Parameter user hinzugefügt worden, der vom Typ User ist (mit Mapped["User"]) und die der Klasse User entsprechende Beziehung angibt (nämlich relationship(back_populates="todos")).

Um den Bezug zwischen den Klassen User und Todo auch zu speichern, wird der Fremdschlüssel user_id als weiterer Parameter in der Klasse Todo aufgenommen. Da hier tatsächlich Daten gespeichert werden, wird wieder die Funktion mapped_column() verwendet, und die Tatsache, dass es sich hierbei um einen Fremdschlüssel handelt mit ForeignKey("users_table.id") angegeben. Der Fremdschlüssel ist hier also in der Tabelle users_table die Spalte id (bzw. der Parameter id der Klasse User). Da jedes Todo-Element auch einem Benutzer zugewiesen sein muss, ist nullable=False gesetzt.

Nun kann man sowohl in der Klasse User als auch in der Klasse Todo auf die Parameter der jeweils anderen Klasse zugreifen, ohne sich Gedanken darüber machen zu müssen, wie die entsprechenden Daten aus der Datenbank ausgelesen werden können. Dies ist im vorstehenden Beispiel gut zu sehen: z.B. in der Property open_todos der Klasse User, wo auf den Parameter done_date von Todo zugegriffen wird; oder in der Methode __repr__(self) der Klasse Todo, wo auf den Parameter name von User zugegriffen wird.

Die Ausgabe in der Befehlszeile ist demnach deutlich informativer, aber der dafür benötigte Code bleibt weiterhin gut verständlich:

[<User(id=1, name=Max, todos=1/2)>, <User(id=2, name=Kathi, todos=0/0)>]
[<Todo(id=1, desc=Write code, user=Max, done_date=2024-10-12 18:03:16.312209)>, <Todo(id=2, desc=Tidy your room, user=Max, done_date=None)>]

Core

Anstatt Klassen zu definieren, die als Modelle für Tabellen einer Datenbank dienen, kann man die Tabellen in SQLAlchemy auch auf eine klassische Art, bzw. direkt erstellen.

Anstelle der Klasse Base braucht man ein Objekt, dass die Metadaten und das Schema verwaltet:

from sqlalchemy import MetaData()

metadata = MetaData()

Nun kann man die Tabellen als Variablen definieren, und das Metadaten-Objekt beim Erstellen übergeben, sodass das Metadaten-Objekt die Tabellen mit seiner Methode create_all() erstellen kann:

from sqlalchemy import (
    create_engine,
    ForeignKey,
    MetaData,
    Table,
    Column,
    String,
    Integer,
    DateTime,
)

users_table = Table(
    "users_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False, unique=True),
)

todos_table = Table(
    "todos_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("description", String, nullable=False),
    Column("done_date", DateTime, nullable=True),
    Column("user_id", Integer, ForeignKey("users_table.id"), nullable=False),
)

def main() -> None:
    connection = engine.connect()
    metadata.create_all(engine)

if __name__ == "__main__":
    main()

Wie man sieht, muss hier eine Verbindung zur Datenbank-Engine erstellt werden, sodass das Metadaten-Objekt die Anweisungen zum Erstellen der Tabellen an die Datenbank übergeben kann. Jetzt sind zwar alle Tabellen erstellt, doch gibt es keine Funktionen, mit denen Daten erzeugt, gespeichert und ausgelesen werden können. Diese muss man nun selber definieren:

from datetime import datetime

def insert_user(name: str) -> None:
    query = users_table.insert().values(name=name)
    connection.execute(query)

def select_user(name: str) -> dict | None:
    query = users_table.select().where(users_table.c.name == name)
    result = connection.execute(query)
    d = result.fetchone()
    if d is None:
        return None
    return dict(zip((c.name for c in users_table.c), d))

def list_users() -> list:
    query = users_table.select()
    result = connection.execute(query)
    return [dict(zip((c.name for c in users_table.c), d)) for d in result.fetchall()]

def insert_todo(description: str, user_id: int, done_date: datetime = None) -> None:
    query = todos_table.insert().values(
        description=description, user_id=user_id, done_date=done_date
    )
    connection.execute(query)

def select_todo(id: int) -> dict | None:
    query = todos_table.select().where(todos_table.c.id == id)
    result = connection.execute(query)
    d = result.fetchone()
    if d is None:
        return None
    return dict(zip((c.name for c in todos_table.c), d))

def list_todos() -> list:
    query = todos_table.select()
    result = connection.execute(query)
    return [dict(zip((c.name for c in todos_table.c), d)) for d in result.fetchall()]

Dies ist deutlich aufwendiger als die Herangehensweise mit ORM. Allerdings hat man mehr Kontrolle über die Abläufe.

Anzumerken ist in dem vorstehenden Code, dass mit (c.name for c in _TABLE_NAME_.c) eine Liste der Spaltennamen erstellt wird, die dann mit dict(zip(a, b)) an ein Dictionary als Schlüsselworte übergeben wird. Das ist ein kleiner syntaktischer Hack, mit dem die Ergebnisse einer Abfrage überschaubar zurückgegeben werden können.

Mit dem vorstehenden Code wird auch der Fremdschlüssel users_table.id als die Tabellenspalte user_id in todos_table gespeichert. Nur muss man hier, um auf den entsprechenden Benutzer zugreifen zu können, auf Grundlage des in todos_table gespeicherten user_id eine erneute Abfrage starten, um den Eintrag des Benutzers aus users_table auszulesen.

Zudem greift jede Abfrage immer auf die Verbindung (also connection) zurück, was bedeutet, dass diese jeder Abfrage-Funktion zur Verfügung stehen muss. Der Einfachheit halber ist die Verbindung im folgenden Code global definiert, doch diese Herangehensweise ist aus Sicherheitsgründen nicht empfehlenswert:

from sqlalchemy import (
    create_engine,
    ForeignKey,
    MetaData,
    Table,
    Column,
    String,
    Integer,
    DateTime,
)

from datetime import datetime

engine = create_engine("sqlite:///:memory:")
connection = engine.connect()

metadata = MetaData()

users_table = Table(
    "users_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False, unique=True),
)

def insert_user(name: str) -> None:
    query = users_table.insert().values(name=name)
    connection.execute(query)

def select_user(name: str) -> dict | None:
    query = users_table.select().where(users_table.c.name == name)
    result = connection.execute(query)
    d = result.fetchone()
    if d is None:
        return None
    return dict(zip((c.name for c in users_table.c), d))

def list_users() -> list:
    query = users_table.select()
    result = connection.execute(query)
    return [dict(zip((c.name for c in users_table.c), d)) for d in result.fetchall()]

todos_table = Table(
    "todos_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("description", String, nullable=False),
    Column("done_date", DateTime, nullable=True),
    Column("user_id", Integer, ForeignKey("users_table.id"), nullable=False),
)

def insert_todo(description: str, user_id: int, done_date: datetime = None) -> None:
    query = todos_table.insert().values(
        description=description, user_id=user_id, done_date=done_date
    )
    connection.execute(query)

def select_todo(id: int) -> dict | None:
    query = todos_table.select().where(todos_table.c.id == id)
    result = connection.execute(query)
    d = result.fetchone()
    if d is None:
        return None
    return dict(zip((c.name for c in todos_table.c), d))

def list_todos() -> list:
    query = todos_table.select()
    result = connection.execute(query)
    return [dict(zip((c.name for c in todos_table.c), d)) for d in result.fetchall()]

def main() -> None:
    metadata.create_all(engine)

    insert_user("Max")
    insert_user("Kathi")

    max = select_user("Max")

    insert_todo("Write code", max["id"], datetime.now())
    insert_todo("Tidy your room", max["id"])

    print(list_users())
    print(list_todos())

    connection.close()

if __name__ == "__main__":
    main()

Die Ausgabe in der Befehlszeile zeigt die gewünschten Informationen an, ist aber im Vergleich zu ORM unvollständiger und (meiner Ansicht nach) mit aufwändigerem Code erzeugt worden:

[{'id': 1, 'name': 'Max'}, {'id': 2, 'name': 'Kathi'}]
[{'id': 1, 'description': 'Write code', 'done_date': datetime.datetime(2024, 10, 12, 10, 4, 23, 249914), 'user_id': 1}, {'id': 2, 'description': 'Tidy your room', 'done_date': None, 'user_id': 1}]