Mehrere Modelle mit FastAPI

Wir haben dasselbe Hero-Modell verwendet, um das Schema der Daten zu deklarieren, die wir in der API empfangen, das Tabellenmodell in der Datenbank und das Schema der Daten, die wir in Antworten zurücksenden.

Aber in den meisten Fällen gibt es geringfügige Unterschiede. Lassen Sie uns mehrere Modelle verwenden, um dies zu lösen.

Hier sehen Sie die wichtigste und größte Funktion von SQLModel. 😎

Überprüfung des Erstellungsschemas

Beginnen wir mit der Überprüfung der automatisch generierten Schemas aus der Doku-UI.

Für die Eingabe haben wir

Wenn wir darauf achten, zeigt dies, dass der Client eine id im JSON-Body der Anfrage senden *könnte*.

Das bedeutet, dass der Client versuchen könnte, dieselbe ID zu verwenden, die bereits in der Datenbank vorhanden ist, um einen weiteren Helden zu erstellen.

Das wollen wir nicht.

Wir möchten, dass der Client nur die Daten sendet, die zur Erstellung eines neuen Helden benötigt werden.

• name
• secret_name
• Optionale age

Und wir möchten, dass die id automatisch von der Datenbank generiert wird, daher möchten wir nicht, dass der Client sie sendet.

Wir werden gleich sehen, wie wir das beheben können.

Überprüfung des Antwortschemas

Überprüfen wir nun das Schema der Antwort, die wir in der Doku-UI an den Client zurücksenden.

Wenn Sie auf den kleinen Tab Schema anstelle von Example Value klicken, sehen Sie etwas wie dies:

Sehen wir uns die Details an.

Die Felder mit einem roten Sternchen (*) sind "erforderlich".

Das bedeutet, dass unsere API-Anwendung diese Felder in der Antwort zurückgeben muss.

• name
• secret_name

Das age ist optional, wir müssen es nicht zurückgeben, oder es könnte None sein (oder null in JSON), aber das name und das secret_name sind erforderlich.

Hier ist die seltsame Sache, die id scheint derzeit auch "optional" zu sein. 🤔

Das liegt daran, dass wir in unserer SQLModel-Klasse die id mit einem Standardwert von = None deklarieren, da sie im Speicher None sein könnte, bis wir sie in der Datenbank speichern und schließlich die tatsächliche ID erhalten.

Aber in den Antworten senden wir immer ein Modell aus der Datenbank, daher **hat es immer eine ID**. Die id in den Antworten kann also als erforderlich deklariert werden.

Das bedeutet, dass unsere Anwendung den Clients das Versprechen macht, dass, wenn sie einen Helden sendet, sie definitiv eine id mit einem Wert hat, sie wird nicht None sein.

Warum ist ein Vertrag für Antworten wichtig?

Das ultimative Ziel einer API ist, dass einige Clients sie verwenden.

Die Clients könnten eine Frontend-Anwendung, ein Kommandozeilenprogramm, eine grafische Benutzeroberfläche, eine mobile Anwendung, eine andere Backend-Anwendung usw. sein.

Und der Code, den diese Clients schreiben, hängt davon ab, was unsere API ihnen sagt, dass sie senden müssen und was sie erwarten können zu erhalten.

Wenn beide Seiten sehr klar sind, wird die Interaktion mit der API viel einfacher.

Und in den meisten Fällen wird der Entwickler des Clients für diese API **auch Sie selbst sein**, also **tun Sie Ihrem zukünftigen Ich einen Gefallen**, indem Sie diese Schemas für Anfragen und Antworten deklarieren. 😉

Warum sind erforderliche IDs wichtig?

Was ist das Problem damit, ein id-Feld als "optional" zu kennzeichnen in einer Antwort, wenn es in Wirklichkeit immer verfügbar (erforderlich) ist?

Zum Beispiel hätten **automatisch generierte Clients** in anderen Sprachen (oder auch in Python) eine Deklaration, dass dieses Feld id optional ist.

Und dann müssten die Entwickler, die diese Clients in ihren Sprachen verwenden, ständig in ihrem gesamten Code prüfen, ob die id nicht None ist, bevor sie sie irgendwo verwenden.

Das sind viele unnötige Prüfungen und **unnötiger Code**, der hätte eingespart werden können, indem das Schema richtig deklariert wurde. 😔

Es wäre für diesen Code viel einfacher zu wissen, dass die id aus einer Antwort erforderlich ist und **immer einen Wert haben wird**.

Das beheben wir auch noch. 🤓

Mehrere Helden-Schemas

Wir möchten also unser Hero-Modell haben, das die **Daten in der Datenbank** deklariert

• id, optional bei Erstellung, erforderlich in der Datenbank
• name, erforderlich
• secret_name, erforderlich
• age, optional

Aber wir möchten auch ein HeroCreate für die Daten haben, die wir beim **Erstellen** eines neuen Helden empfangen möchten, was fast dieselben Daten wie Hero sind, außer der id, da diese automatisch von der Datenbank erstellt wird.

• name, erforderlich
• secret_name, erforderlich
• age, optional

Und wir möchten ein HeroPublic mit dem id-Feld haben, aber diesmal mit dem Typ id: int, anstatt id: int | None, um klarzustellen, dass es immer eine int in Antworten geben wird, die **von den Clients gelesen** werden.

• id, erforderlich
• name, erforderlich
• secret_name, erforderlich
• age, optional

Mehrere Modelle mit doppelten Feldern

Der einfachste Weg, dies zu lösen, könnte darin bestehen, **mehrere Modelle** zu erstellen, jedes mit allen entsprechenden Feldern.

Python 3.10+

# Code above omitted 👆

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Hier ist das wichtige Detail und wahrscheinlich die wichtigste Funktion von SQLModel: Nur Hero ist mit table = True deklariert.

Das bedeutet, dass die Klasse Hero eine **Tabelle** in der Datenbank repräsentiert. Sie ist sowohl ein Pydantic-Modell als auch ein SQLAlchemy-Modell.

Aber HeroCreate und HeroPublic haben kein table = True. Sie sind nur **Datenmodelle**, sie sind nur **Pydantic**-Modelle. Sie werden nicht mit der Datenbank verwendet, sondern nur zur Deklaration von Datenschemata für die API (oder für andere Zwecke).

Das bedeutet auch, dass SQLModel.metadata.create_all() keine Tabellen in der Datenbank für HeroCreate und HeroPublic erstellt, da sie kein table = True haben, was genau das ist, was wir wollen. 🚀

Tipp

Wir werden diesen Code verbessern, um die Duplizierung von Feldern zu vermeiden, aber vorerst können wir mit diesen Modellen weiterlernen.

Verwenden von mehreren Modellen zum Erstellen eines Helden

Lassen Sie uns nun sehen, wie wir diese neuen Modelle in der FastAPI-Anwendung verwenden.

Überprüfen wir zunächst, wie der Prozess zur Erstellung eines Helden jetzt aussieht.

Python 3.10+

# Code above omitted 👆

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Sehen wir uns das im Detail an.

Jetzt verwenden wir die Typannotation HeroCreate für die JSON-Daten der Anfrage im Parameter hero der **Pfadoperationsfunktion**.

Python 3.10+

# Code above omitted 👆

def create_hero(hero: HeroCreate):

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Dann erstellen wir einen neuen Hero (dies ist das tatsächliche **Tabellenmodell**, das Dinge in der Datenbank speichert) mit Hero.model_validate().

Die Methode .model_validate() liest Daten aus einem anderen Objekt mit Attributen (oder einem Dict) und erstellt eine neue Instanz dieser Klasse, in diesem Fall Hero.

In diesem Fall haben wir eine HeroCreate-Instanz in der Variablen hero. Dies ist ein Objekt mit Attributen, daher verwenden wir .model_validate(), um diese Attribute zu lesen.

Tipp

In Versionen von SQLModel vor 0.0.14 würden Sie die Methode .from_orm() verwenden, aber sie ist jetzt veraltet und Sie sollten stattdessen .model_validate() verwenden.

Wir können jetzt eine neue Hero-Instanz (die für die Datenbank) erstellen und sie aus den Daten in der hero-Variablen, die die HeroCreate-Instanz ist, die wir von der Anfrage erhalten haben, in die Variable db_hero einfügen.

Python 3.10+

# Code above omitted 👆

        db_hero = Hero.model_validate(hero)

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Dann fügen wir sie einfach dem **Session** hinzu, machen commit und refresh sie, und schließlich geben wir dieselbe db_hero-Variable zurück, die die gerade aktualisierte Hero-Instanz enthält.

Da sie gerade aktualisiert wurde, hat sie das Feld id, das mit einer neuen ID aus der Datenbank gesetzt ist.

Und jetzt, da wir sie zurückgeben, validiert FastAPI die Daten mit dem response_model, das ein HeroPublic ist.

Python 3.10+

# Code above omitted 👆

@app.post("/heroes/", response_model=HeroPublic)

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class HeroCreate(SQLModel):
    name: str
    secret_name: str
    age: Optional[int] = None


class HeroPublic(SQLModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Dies validiert, dass alle Daten, die wir versprochen haben, vorhanden sind, und entfernt alle Daten, die wir nicht deklariert haben.

Tipp

Diese Filterung könnte sehr wichtig sein und eine sehr gute Sicherheitsfunktion darstellen, zum Beispiel um sicherzustellen, dass Sie private Daten, gehashte Passwörter usw. filtern.

Sie können mehr darüber in den FastAPI-Dokumenten zu Response Model lesen.

Insbesondere stellt dies sicher, dass die id vorhanden ist und dass es sich tatsächlich um eine Ganzzahl handelt (und nicht um None).

Gemeinsame Felder

Wenn wir jedoch genau hinschauen, sehen wir, dass diese Modelle viele **duplizierte Informationen** enthalten.

Alle **die 3 Modelle** deklarieren, dass sie einige **gemeinsame Felder** teilen, die genau gleich aussehen.

• name, erforderlich
• secret_name, erforderlich
• age, optional

Und dann deklarieren sie andere Felder mit einigen Unterschieden (in diesem Fall nur bezüglich der id).

Wir wollen **duplizierte Informationen vermeiden**, wenn möglich.

Dies ist wichtig, wenn wir zum Beispiel in Zukunft **den Code refaktorisieren** und ein Feld (Spalte) umbenennen. Zum Beispiel von secret_name zu secret_identity.

Wenn wir das in mehreren Modellen dupliziert haben, könnten wir leicht vergessen, eines davon zu aktualisieren. Aber wenn wir **Duplikation vermeiden**, gibt es nur einen Ort, der aktualisiert werden müsste. ✨

Verbessern wir das jetzt. 🤓

Mehrere Modelle mit Vererbung

Und hier ist es, Sie haben die größte Funktion von SQLModel gefunden. 💎

Jedes dieser Modelle ist nur ein **Datenmodell** oder sowohl ein Datenmodell als auch ein **Tabellenmodell**.

Daher ist es möglich, Modelle mit SQLModel zu erstellen, die keine Tabellen in der Datenbank darstellen.

Darüber hinaus können wir Vererbung verwenden, um duplizierte Informationen in diesen Modellen zu vermeiden.

Wir sehen von oben, dass sie alle einige **Basis**felder gemeinsam haben.

• name, erforderlich
• secret_name, erforderlich
• age, optional

Erstellen wir also ein **Basis**modell HeroBase, von dem die anderen erben können.

Python 3.10+

# Code above omitted 👆

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Wie Sie sehen können, ist dies *kein* **Tabellenmodell**, es hat nicht die table = True-Konfiguration.

Aber jetzt können wir die **anderen Modelle erstellen, die davon erben**. Sie werden all diese Felder gemeinsam nutzen, als ob sie sie deklariert hätten.

Das Hero Tabellenmodell

Beginnen wir mit dem einzigen **Tabellenmodell**, dem Hero.

Python 3.10+

# Code above omitted 👆

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Beachten Sie, dass Hero jetzt nicht von SQLModel erbt, sondern von HeroBase.

Und jetzt deklarieren wir nur ein einziges Feld direkt, die id, die hier int | None ist und ein primary_key ist.

Und obwohl wir die anderen Felder **nicht explizit** deklarieren, sind sie, da sie vererbt werden, auch Teil dieses Hero-Modells.

Und natürlich werden all diese Felder in den Spalten für die resultierende hero-Tabelle in der Datenbank enthalten sein.

Und diese vererbten Felder werden auch in der **Autovervollständigung** und den **Inline-Fehlern** in Editoren usw. enthalten sein.

Spalten und Vererbung mit mehreren Modellen

Beachten Sie, dass das Elternmodell HeroBase kein **Tabellenmodell** ist, aber wir können dennoch name und age mit Field(index=True) deklarieren.

Python 3.10+

# Code above omitted 👆

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Dies hat keine Auswirkungen auf dieses übergeordnete **Datenmodell** HeroBase.

Aber sobald das Kindmodell Hero (das tatsächliche **Tabellenmodell**) diese Felder erbt, wird es diese Feldkonfigurationen verwenden, um die Indizes bei der Erstellung der Tabellen in der Datenbank zu erstellen.

Das HeroCreate Datenmodell

Sehen wir uns nun das HeroCreate-Modell an, das zur Definition der Daten verwendet wird, die wir bei der Erstellung eines neuen Helden in der API empfangen möchten.

Das ist ein lustiges.

Python 3.10+

# Code above omitted 👆

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Was passiert hier?

Die Felder, die wir zur Erstellung benötigen, sind **genau dieselben** wie die im HeroBase-Modell. Wir müssen also nichts hinzufügen.

Und da wir den leeren Raum beim Erstellen einer neuen Klasse nicht lassen können, aber kein Feld hinzufügen möchten, verwenden wir einfach pass.

Das bedeutet, dass es in dieser Klasse nichts Besonderes gibt, außer dass sie HeroCreate heißt und von HeroBase erbt.

Alternativ könnten wir HeroBase direkt im API-Code anstelle von HeroCreate verwenden, aber es würde in der automatischen Doku-UI mit dem Namen "HeroBase" angezeigt werden, was für Clients **verwirrend** sein könnte. Stattdessen ist "HeroCreate" etwas expliziter, wofür es gedacht ist.

Darüber hinaus könnten wir in Zukunft leicht entscheiden, dass wir **mehr Daten** beim Erstellen eines neuen Helden über die Daten in HeroBase hinaus empfangen möchten (z. B. ein Passwort), und jetzt haben wir bereits die Klasse, um diese zusätzlichen Felder unterzubringen.

Das HeroPublic Datenmodell

Überprüfen wir nun das HeroPublic-Modell.

Dies deklariert lediglich, dass das Feld id beim Lesen eines Helden aus der API erforderlich ist, da ein aus der API gelesener Held aus der Datenbank stammt und in der Datenbank immer eine ID haben wird.

Python 3.10+

# Code above omitted 👆

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

Überprüfung der aktualisierten Doku-UI

Der FastAPI-Code ist immer noch derselbe wie oben, wir verwenden immer noch Hero, HeroCreate und HeroPublic. Aber jetzt definieren wir sie auf intelligentere Weise mit Vererbung.

Wir können also sofort zur Doku-UI springen und sehen, wie sie mit den aktualisierten Daten aussehen.

Doku-UI zum Erstellen eines Helden

Sehen wir uns die neue UI zum Erstellen eines Helden an.

Schön! Es wird jetzt angezeigt, dass wir zum Erstellen eines Helden nur name, secret_name und optional age übergeben müssen.

Wir übergeben keine id mehr.

Doku-UI mit Helden-Antworten

Jetzt können wir ein wenig nach unten scrollen, um das Antwortschema zu sehen.

Wir sehen jetzt, dass id ein erforderliches Feld ist, es hat ein rotes Sternchen (*).

Und wenn wir das Schema für die *Pfadoperation* **Helden lesen** überprüfen, wird auch das aktualisierte Schema angezeigt.

Vererbung und Tabellenmodelle

Wir haben gerade gesehen, wie leistungsfähig die Vererbung dieser Modelle sein kann.

Dies ist ein sehr einfaches Beispiel und es mag ein wenig... meh. 😅

Aber stellen Sie sich nun vor, Ihre Tabelle hat **10 oder 20 Spalten**. Und Sie müssen all diese Informationen für alle Ihre **Datenmodelle** duplizieren... dann wird es deutlicher, warum es ziemlich nützlich ist, all diese Informationsduplikation mit Vererbung vermeiden zu können.

Nun, das sieht wahrscheinlich so flexibel aus, dass es nicht offensichtlich ist, **wann Vererbung verwendet werden soll** und wofür.

Hier sind ein paar Faustregeln, die Ihnen helfen können.

Nur von Datenmodellen erben

Erben Sie nur von **Datenmodellen**, erben Sie nicht von **Tabellenmodellen**.

Das hilft Ihnen, Verwirrung zu vermeiden, und es wird keinen Grund geben, von einem **Tabellenmodell** erben zu müssen.

Wenn Sie das Gefühl haben, von einem **Tabellenmodell** erben zu müssen, dann erstellen Sie stattdessen eine **Basisklasse**, die nur ein **Datenmodell** ist und all diese Felder enthält, wie HeroBase.

Und erben Sie dann von dieser **Basisklasse**, die nur ein **Datenmodell** ist, für jedes andere **Datenmodell** und für das **Tabellenmodell**.

Vermeiden Sie Duplizierung - Halten Sie es einfach

Es mag sich anfühlen, als ob Sie einen tiefgreifenden Grund hätten, von einem Modell oder einem anderen zu erben, weil sie "auf mystische Weise" verschiedene Konzepte trennen... oder so etwas.

In einigen Fällen gibt es **einfache Trennungen**, die Sie verwenden können, wie Modelle zur Erstellung, zum Lesen, Aktualisieren usw. von Daten. Wenn das schnell und offensichtlich ist, schön, nutzen Sie es. 💯

Andernfalls machen Sie sich nicht zu viele Gedanken über tiefgreifende konzeptionelle Gründe, Modelle zu trennen, versuchen Sie einfach, **Duplikation zu vermeiden** und **den Code einfach zu halten**, damit er nachvollziehbar ist.

Wenn Sie feststellen, dass es viel **Überschneidung** zwischen zwei Modellen gibt, dann können Sie wahrscheinlich **einige dieser Duplikation mit einem Basismodell vermeiden**.

Aber wenn Sie am Ende einen verrückten Baum von Modellen mit Vererbung haben, um einige Duplikationen zu vermeiden, dann ist es vielleicht **einfacher**, einige dieser Felder einfach zu duplizieren, und das könnte leichter nachvollziehbar und wartbar sein.

Machen Sie, was am einfachsten ist, um darüber zu **nachzudenken**, damit zu **programmieren**, es zu **warten** und es in Zukunft zu **refaktorisieren**. 🤓

Denken Sie daran, dass Vererbung, genauso wie SQLModel und alles andere, nur Werkzeuge sind, die Ihnen helfen, **produktiver zu sein**. Das ist eines ihrer Hauptziele. Wenn etwas nicht dabei hilft (z. B. zu viel Duplizierung, zu viel Komplexität), ändern Sie es. 🚀

Zusammenfassung

Sie können SQLModel verwenden, um mehrere Modelle zu deklarieren.

• Einige Modelle können nur **Datenmodelle** sein. Sie werden auch **Pydantic**-Modelle sein.
• Und einige können *auch* **Tabellenmodelle** sein (abgesehen davon, dass sie bereits **Datenmodelle** sind), indem sie die Konfiguration table = True haben. Sie werden auch **Pydantic**-Modelle und **SQLAlchemy**-Modelle sein.

Nur die **Tabellenmodelle** erstellen Tabellen in der Datenbank.

Sie können also alle anderen **Datenmodelle** verwenden, um das Schema der Daten für Ihre Anwendung zu validieren, zu konvertieren, zu filtern und zu dokumentieren. ✨

Sie können Vererbung verwenden, um **Informations- und Code-Duplikation zu vermeiden**. 😎

Und Sie können all diese Modelle direkt mit **FastAPI** verwenden. 🚀