Update mit zusätzlichen Daten (gehashte Passwörter) mit FastAPI¶
Im vorherigen Kapitel habe ich Ihnen erklärt, wie Sie Daten in der Datenbank aus Eingabedaten aktualisieren, die von einer FastAPI *Pfadoperation* kommen.
Jetzt erkläre ich Ihnen, wie Sie zusätzliche Daten, zusätzlich zu den Eingabedaten, beim Aktualisieren oder Erstellen eines Modellobjekts hinzufügen.
Dies ist besonders nützlich, wenn Sie einige Daten in Ihrem Code generieren müssen, die nicht vom Client kommen, aber Sie müssen sie in der Datenbank speichern. Zum Beispiel, um ein gehashtes Passwort zu speichern.
Passwort-Hashing¶
Stellen wir uns vor, jeder Held in unserem System hat auch ein Passwort.
Wir sollten das Passwort niemals im Klartext in der Datenbank speichern, wir sollten nur eine gehashte Version davon speichern.
"Hashing" bedeutet, einige Inhalte (in diesem Fall ein Passwort) in eine Byte-Sequenz (nur eine Zeichenkette) umzuwandeln, die wie Kauderwelsch aussieht.
Immer wenn Sie genau denselben Inhalt (genau dasselbe Passwort) eingeben, erhalten Sie genau dasselbe Kauderwelsch.
Aber Sie können nicht vom Kauderwelsch zurück zum Passwort konvertieren.
Warum Passwort-Hashing verwenden?¶
Wenn Ihre Datenbank gestohlen wird, hat der Dieb nicht die Klartext-Passwörter Ihrer Benutzer, sondern nur die Hashes.
Der Dieb kann also nicht versuchen, dieses Passwort in einem anderen System zu verwenden (da viele Benutzer überall dasselbe Passwort verwenden, wäre das gefährlich).
Tipp
Sie könnten passlib zum Hashen von Passwörtern verwenden.
In diesem Beispiel verwenden wir eine gefälschte Hashing-Funktion, um uns auf die Datenänderungen zu konzentrieren. 🤡
Modelle mit zusätzlichen Daten aktualisieren¶
Das Hero Tabellenmodell wird nun ein neues Feld hashed_password speichern.
Und die Datenmodelle für HeroCreate und HeroUpdate werden ebenfalls ein neues Feld password haben, das das vom Client gesendete Klartext-Passwort enthält.
# 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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | None = None
# Code below omitted 👇
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
Wenn ein Client einen neuen Helden erstellt, sendet er das password im Request Body.
Und wenn er einen Helden aktualisiert, könnte er auch das password im Request Body senden, um es zu aktualisieren.
Das Passwort hashen¶
Die App empfängt die Daten vom Client mithilfe des HeroCreate Modells.
Dieses enthält das Feld password mit dem Klartext-Passwort, und das können wir nicht verwenden. Also müssen wir daraus einen Hash generieren.
# Code above omitted 👆
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
# Code here omitted 👈
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
# Code below omitted 👇
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
Ein Objekt mit zusätzlichen Daten erstellen¶
Jetzt müssen wir den Datenbank-Helden erstellen.
In früheren Beispielen haben wir etwas wie das hier verwendet:
db_hero = Hero.model_validate(hero)
Dies erstellt ein Hero (was ein *Tabellenmodell* ist) -Objekt aus dem HeroCreate (was ein *Datenmodell* ist) -Objekt, das wir in der Anfrage erhalten haben.
Und das ist alles gut... aber da Hero kein Feld password hat, wird es nicht aus dem Objekt HeroCreate extrahiert, das es hat.
Hero hat tatsächlich ein hashed_password, aber wir stellen es nicht zur Verfügung. Wir müssen einen Weg finden, es bereitzustellen...
Dictionary-Update¶
Halten wir kurz inne, um dies zu überprüfen. Wenn wir mit Dictionaries arbeiten, gibt es eine Möglichkeit, ein Dictionary mit zusätzlichen Daten aus einem anderen Dictionary zu aktualisieren, so etwas wie:
db_user_dict = {
"name": "Deadpond",
"secret_name": "Dive Wilson",
"age": None,
}
hashed_password = "fakehashedpassword"
extra_data = {
"hashed_password": hashed_password,
"age": 32,
}
db_user_dict.update(extra_data)
print(db_user_dict)
# {
# "name": "Deadpond",
# "secret_name": "Dive Wilson",
# "age": 32,
# "hashed_password": "fakehashedpassword",
# }
Diese update-Methode ermöglicht es uns, Dinge im ursprünglichen Dictionary mit den Daten aus einem anderen Dictionary hinzuzufügen und zu überschreiben.
Jetzt hat db_user_dict das aktualisierte Feld age mit 32 anstelle von None und, was noch wichtiger ist, es hat das neue Feld hashed_password.
Ein Modellobjekt mit zusätzlichen Daten erstellen¶
Ähnlich wie Dictionaries eine update-Methode haben, verfügen SQLModel Modelle über einen Parameter update in Hero.model_validate(), der ein Dictionary mit zusätzlichen Daten oder Daten, die Vorrang haben sollen, entgegennimmt.
# Code above omitted 👆
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
# Code below omitted 👇
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
Jetzt extrahiert db_hero (was ein *Tabellenmodell* Hero ist) seine Werte aus hero (was ein *Datenmodell* HeroCreate ist), und dann aktualisiert es seine Werte mit den zusätzlichen Daten aus dem Dictionary extra_data.
Es werden nur die in Hero definierten Felder übernommen, daher wird das password aus HeroCreate nicht übernommen. Und es übernimmt seine Werte auch aus dem an den Parameter update übergebenen Dictionary, in diesem Fall dem hashed_password.
Wenn ein Feld sowohl in hero als auch in extra_data vorhanden ist, haben die Werte aus extra_data, die an update übergeben werden, Vorrang.
Update mit zusätzlichen Daten¶
Nehmen wir nun an, wir möchten einen Helden aktualisieren, der bereits in der Datenbank vorhanden ist.
Auf die gleiche Weise wie zuvor, um das Entfernen vorhandener Daten zu vermeiden, verwenden wir exclude_unset=True beim Aufruf von hero.model_dump(), um ein Dictionary mit nur den vom Client gesendeten Daten zu erhalten.
# Code above omitted 👆
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
# Code below omitted 👇
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
Dieses hero_data-Dictionary könnte ein password enthalten. Wir müssen es überprüfen und wenn es vorhanden ist, müssen wir das hashed_password generieren.
Dann können wir dieses hashed_password in ein Dictionary einfügen.
Und dann können wir das db_hero-Objekt mit der Methode db_hero.sqlmodel_update() aktualisieren.
Sie nimmt ein Modellobjekt oder ein Dictionary mit den zu aktualisierenden Daten entgegen und auch ein zusätzliches update-Argument mit zusätzlichen Daten.
# Code above omitted 👆
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
👀 Vorschau der vollständigen Datei
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: str | None = None
secret_name: str | None = None
age: int | None = None
password: str | 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
🤓 Andere Versionen und Varianten
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Query
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)
hashed_password: str = Field()
class HeroCreate(HeroBase):
password: str
class HeroPublic(HeroBase):
id: int
class HeroUpdate(SQLModel):
name: Optional[str] = None
secret_name: Optional[str] = None
age: Optional[int] = None
password: Optional[str] = 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)
def hash_password(password: str) -> str:
# Use something like passlib here
return f"not really hashed {password} hehehe"
app = FastAPI()
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
hashed_password = hash_password(hero.password)
with Session(engine) as session:
extra_data = {"hashed_password": hashed_password}
db_hero = Hero.model_validate(hero, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
with Session(engine) as session:
heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
return heroes
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
with Session(engine) as session:
hero = session.get(Hero, hero_id)
if not hero:
raise HTTPException(status_code=404, detail="Hero not found")
return hero
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate):
with Session(engine) as session:
db_hero = session.get(Hero, hero_id)
if not db_hero:
raise HTTPException(status_code=404, detail="Hero not found")
hero_data = hero.model_dump(exclude_unset=True)
extra_data = {}
if "password" in hero_data:
password = hero_data["password"]
hashed_password = hash_password(password)
extra_data["hashed_password"] = hashed_password
db_hero.sqlmodel_update(hero_data, update=extra_data)
session.add(db_hero)
session.commit()
session.refresh(db_hero)
return db_hero
Tipp
Die Methode db_hero.sqlmodel_update() wurde in SQLModel 0.0.16 hinzugefügt. 😎
Zusammenfassung¶
Sie können den Parameter update in Hero.model_validate() verwenden, um zusätzliche Daten beim Erstellen eines neuen Objekts bereitzustellen, und Hero.sqlmodel_update(), um zusätzliche Daten beim Aktualisieren eines vorhandenen Objekts bereitzustellen. 🤓