--- title: SQLAlchemy url: 'https://gerfficient.com/de/home/sqlalchemy' markdown: 'https://gerfficient.com/de/home/sqlalchemy.md' lang: de date: '2024-10-20' description: '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 Date…' --- # SQLAlchemy ### 20. Okt. 2024 ## 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](https://docs.python.org/3/library/pickle.html)). 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](https://docs.python.org/3/library/sqlite3.html)). Doch mangelt es diesen Modulen an der Benutzerfreundlichkeit, da man jede Datenabfrage selbst als SQL-Abfrage formulieren und auswerten muss. Das Modul [SQLAlchemy](https://www.sqlalchemy.org) 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: ```sh 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: ```python 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: ```python 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](https://peps.python.org/pep-0484/) eingesetzt: ```Python 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"" 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"" ``` 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: ```Python 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: ``` [, ] [, ] ``` 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: ```Python ... 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 `` zu `` 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: ```Python 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"" @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"" 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: ```Python [, ] [, ] ``` ### 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: ```python 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: ```Python 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: ```Python 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: ```Python 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}] ``` --- ## Navigation - Parent: [Home](https://gerfficient.com/de/index.md) - Previous: [Webpage im Homelab](https://gerfficient.com/de/home/homelab-webpage.md) - Next: [Neuinstallation eines Mac](https://gerfficient.com/de/home/fresh-installation.md)