FastAPIと複数モデル¶
これまで、APIで受信するデータのスキーマ、データベースのテーブルモデル、レスポンスで送り返すデータのスキーマを宣言するために、同じHero
モデルを使用してきました。
しかし、ほとんどの場合、わずかな違いがあります。それを解決するために、複数のモデルを使用してみましょう。
ここでは、**SQLModel**の主要かつ最大の機能を見ていきます。😎
レビュー作成スキーマ¶
ドキュメントUIから自動生成されたスキーマを確認することから始めましょう。
入力として、以下があります。
注意深く見ると、クライアントがリクエストのJSONボディにid
を送信 *できる* ことがわかります。
これは、クライアントがデータベースに既に存在する別のヒーローと同じIDを使用しようとする可能性があることを意味します。
それは私たちが望むことではありません。
クライアントには、新しいヒーローを作成するために必要なデータのみを送信してもらいたいのです。
name
secret_name
- オプションの
age
そして、id
はデータベースによって自動的に生成されるようにしたいので、クライアントに送信する必要はありません。
修正方法はすぐに見ていきます。
レビューレスポンススキーマ¶
次に、ドキュメントUIでクライアントに送り返すレスポンスのスキーマを確認してみましょう。
Example Valueの代わりに小さなタブSchemaをクリックすると、次のようなものが見えます。
詳細を見てみましょう。
赤いアスタリスク(*)が付いているフィールドは「必須」です。
これは、私たちのAPIアプリケーションがレスポンスでこれらのフィールドを返す必要があることを意味します。
name
secret_name
age
はオプションです。返す必要はありませんし、None
(またはJSONではnull
)でも構いませんが、name
とsecret_name
は必須です。
奇妙なことに、id
は現在「オプション」のようです。🤔
これは、**SQLModel**クラスでid
をOptional[int]
で宣言しているためです。データベースに保存するまでメモリ内ではNone
になる可能性があり、最終的に実際のIDを取得します。
しかし、レスポンスでは常にデータベースからのモデルを送信しているので、**常にIDがあります**。そのため、レスポンスのid
は必須として宣言できます。
これは、アプリケーションがクライアントに対して、ヒーローを送信する場合、確実に値を持つid
があり、None
ではないという約束をしていることを意味します。
レスポンスの契約を持つことの重要性¶
APIの究極の目標は、いくつかの**クライアントがそれを利用すること**です。
クライアントは、フロントエンドアプリケーション、コマンドラインプログラム、グラフィカルユーザーインターフェース、モバイルアプリケーション、別のバックエンドアプリケーションなどです。
そして、これらのクライアントが記述するコードは、私たちのAPIが彼らに**送信する必要があるもの**と、**受信できるもの**に依存します。
両方を明確にすることで、APIとのやり取りがはるかに容易になります。
そして、ほとんどの場合、そのAPIのクライアントの開発者も**あなた自身**であるため、リクエストとレスポンスのスキーマを宣言することで、**将来の自分のためにもなります**。😉
必須IDを持つことの重要性¶
では、実際には常に必須であるのに、レスポンスで1つの**id
フィールドが「オプション」としてマークされているとどうなるのでしょうか?
たとえば、他の言語(またはPythonでも)で**自動生成されたクライアント**では、このフィールドid
がオプションであるという宣言があります。
そして、これらのクライアントを自分の言語で使用している開発者は、コードのいたるところでid
がNone
ではないかどうかを常にチェックする必要があります。
スキーマを適切に宣言することで節約できた、多くの不要なチェックと**不要なコード**です。😔
レスポンスからのid
は必須であり、**常に値を持つ**ことをそのコードが知っている方がはるかに簡単です。
それも修正しましょう。🤓
複数のヒーロースキーマ¶
つまり、データベースの**データ**を宣言するHero
モデルが必要です。
id
、作成時はオプション、データベースでは必須name
、必須secret_name
、必須age
、オプション
しかし、新しいヒーローを作成するときに受信したいデータのためのHeroCreate
も必要です。これはHero
とほぼ同じデータですが、id
はデータベースによって自動的に作成されるため、除外されます。
name
、必須secret_name
、必須age
、オプション
そして、id
フィールドを持つHeroPublic
も必要です。ただし、今回はid: Optional[int]
ではなくid: int
でアノテーションして、クライアントから**読み込まれた**レスポンスでは必須であることを明確にします。
id
、必須name
、必須secret_name
、必須age
、オプション
重複フィールドを持つ複数モデル¶
それを解決する最も簡単な方法は、それぞれに対応するフィールドを持つ**複数のモデル**を作成することです。
# This would work, but there's a better option below 🚨
# 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 👇
# This would work, but there's a better option below 🚨
# Code above omitted 👆
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
# Code below omitted 👇
# This would work, but there's a better option below 🚨
# Code above omitted 👆
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
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
重要な詳細、そしておそらく**SQLModel**の最も重要な機能は、Hero
のみがtable = True
で宣言されていることです。
これは、クラスHero
がデータベースの**テーブル**を表していることを意味します。これは**Pydantic**モデルと**SQLAlchemy**モデルの両方です。
しかし、HeroCreate
とHeroPublic
にはtable = True
がありません。これらは**データモデル**のみであり、**Pydantic**モデルのみです。データベースでは使用されませんが、APIのデータスキーマ(またはその他の用途)の宣言のみに使用されます。
これはまた、SQLModel.metadata.create_all()
がHeroCreate
とHeroPublic
のデータベースにテーブルを作成しないことを意味します。これらはtable = True
を持っていないためです。まさに私たちが望むとおりです。🚀
ヒント
フィールドの重複を避けるためにこのコードを改善しますが、今のところこれらのモデルで学習を続けられます。
ヒーローを作成するための複数モデルの使用¶
これらの新しいモデルをFastAPIアプリケーションでどのように使用するかを見てみましょう。
まず、ヒーローを作成するプロセスがどのように変更されたかを確認しましょう。
# 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 👇
# 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 👇
# 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 👇
👀 ファイル全体のプレビュー
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
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
詳細を確認しましょう。
ここでは、**パスオペレーション関数**のhero
パラメータのリクエストJSONデータに型アノテーションHeroCreate
を使用します。
# Code above omitted 👆
def create_hero(hero: HeroCreate):
# Code below omitted 👇
# Code above omitted 👆
def create_hero(hero: HeroCreate):
# Code below omitted 👇
# Code above omitted 👆
def create_hero(hero: HeroCreate):
# Code below omitted 👇
次に、Hero.model_validate()
を使用して、新しいHero
(データベースにデータを保存する実際の**テーブル**モデル)を作成します。
メソッド.model_validate()
は、属性を持つ別のオブジェクト(または辞書)からデータを読み取り、このクラス、この場合はHero
の新しいインスタンスを作成します。
この場合、hero
変数にHeroCreate
インスタンスがあります。これは属性を持つオブジェクトなので、.model_validate()
を使用してその属性を読み取ります。
ヒント
SQLModel の 0.0.14
より前のバージョンでは、.from_orm()
メソッドを使用していましたが、現在は非推奨となっており、代わりに .model_validate()
を使用する必要があります。
リクエストから受け取った HeroCreate
インスタンスである hero
変数のデータから、新しい Hero
インスタンス(データベース用のもの)を作成し、db_hero
変数に格納できるようになりました。
# Code above omitted 👆
db_hero = Hero.model_validate(hero)
# Code below omitted 👇
# Code above omitted 👆
db_hero = Hero.model_validate(hero)
# Code below omitted 👇
# Code above omitted 👆
db_hero = Hero.model_validate(hero)
# Code below omitted 👇
その後、それをセッションに add
し、commit
し、refresh
します。最後に、更新されたばかりの Hero
インスタンスを含む同じ db_hero
変数を返します。
これは更新されたばかりなので、データベースから取得した新しいIDが設定された id
フィールドを持っています。
そして、それを返すことで、FastAPI は response_model
(HeroPublic
)を使用してデータを検証します。
# Code above omitted 👆
@app.post("/heroes/", response_model=HeroPublic)
# Code below omitted 👇
# Code above omitted 👆
@app.post("/heroes/", response_model=HeroPublic)
# Code below omitted 👇
# Code above omitted 👆
@app.post("/heroes/", response_model=HeroPublic)
# Code below omitted 👇
これにより、約束されたすべてのデータが存在することが検証され、宣言されていないデータは削除されます。
ヒント
このフィルタリングは非常に重要であり、非常に優れたセキュリティ機能となり得ます。たとえば、プライベートデータ、ハッシュされたパスワードなどを確実にフィルタリングするために使用できます。
レスポンスモデルに関するFastAPI ドキュメントで、詳細を読むことができます。
特に、id
が存在し、それが実際には整数(None
ではない)であることを確認します。
共有フィールド¶
しかし、よく見ると、これらのモデルには多くの重複した情報があることがわかります。
3つのモデルすべてが、まったく同じに見えるいくつかの共通フィールドを宣言しています。
name
、必須secret_name
、必須age
、オプション
そして、いくつかの違い(この場合は、id
のみ)を持つ他のフィールドを宣言しています。
可能であれば、重複した情報を避ける必要があります。
これは、たとえば、将来、コードをリファクタリングしてフィールド(カラム)の名前を変更する場合(たとえば、secret_name
からsecret_identity
に変更する場合)に重要です。
複数のモデルで重複している場合、更新を忘れる可能性があります。しかし、重複を回避すれば、更新が必要となる場所は1箇所のみです。✨
それでは、それを改善しましょう。🤓
継承による複数のモデル¶
そしてここに、SQLModel の最大の機能があります。💎
これらのモデルのそれぞれは、データモデルのみ、またはデータモデルとテーブルモデルの両方です。
そのため、データベース内のテーブルを表さないSQLModel を使用してモデルを作成できます。
さらに、継承を使用してこれらのモデルの重複情報を回避できます。
上記から、それらすべてがいくつかの基本的なフィールドを共有していることがわかります。
name
、必須secret_name
、必須age
、オプション
そこで、他のモデルが継承できる基本的なモデルHeroBase
を作成しましょう。
# 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 👇
# Code above omitted 👆
class HeroBase(SQLModel):
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
# Code below omitted 👇
# Code above omitted 👆
class HeroBase(SQLModel):
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None, index=True)
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
ご覧のとおり、これはテーブルモデルではありません。table = True
の設定がありません。
しかし、これでそれから継承する他のモデルを作成できます。それらは、宣言されているかのように、これらのフィールドをすべて共有します。
Hero
テーブルモデル¶
唯一のテーブルモデルであるHero
から始めましょう。
# 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 👇
# Code above omitted 👆
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)
# Code below omitted 👇
# Code above omitted 👆
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)
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
Hero
は、SQLModel
ではなくHeroBase
を継承していることに注意してください。
そして、ここではOptional[int]
であり、primary_key
であるid
という単一フィールドのみを直接宣言します。
そして、明示的に他のフィールドを宣言していなくても、継承されているため、それらはこのHero
モデルの一部でもあります。
そしてもちろん、これらのフィールドはすべて、データベース内の結果のhero
テーブルのカラムに含まれます。
そして、これらの継承されたフィールドは、エディタなどの自動補完とインラインエラーにも含まれます。
複数のモデルを使用したカラムと継承¶
親モデルHeroBase
はテーブルモデルではありませんが、それでもField(index=True)
を使用してname
とage
を宣言できます。
# 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 👇
# Code above omitted 👆
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)
# Code below omitted 👇
# Code above omitted 👆
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)
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
これは、この親データモデルHeroBase
には影響しません。
しかし、子モデルHero
(実際のテーブルモデル)がこれらのフィールドを継承すると、データベースにテーブルを作成する際に、これらのフィールド設定を使用してインデックスを作成します。
HeroCreate
データモデル¶
新しいヒーローを作成する際にAPIで受信したいデータを定義するために使用されるHeroCreate
モデルを見てみましょう。
これは楽しいものです。
# 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 👇
# Code above omitted 👆
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
# Code below omitted 👇
# Code above omitted 👆
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
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
ここで何が起こっているのでしょうか?
作成する必要があるフィールドは、HeroBase
モデルのフィールドとまったく同じです。そのため、何も追加する必要はありません。
新しいクラスを作成する際に空のスペースを残すことはできませんが、フィールドを追加したくないため、pass
を使用します。
これは、このクラスには、HeroCreate
という名前であり、HeroBase
を継承しているという事実以外に、特別なものは何もないことを意味します。
代替として、APIコードでHeroCreate
の代わりにHeroBase
を直接使用できますが、自動ドキュメントUIに「HeroBase
」という名前で表示され、クライアントにとって分かりにくい可能性があります。「HeroCreate
」の方が、それが何のためにあるのかをより明確に示しています。
さらに、将来、HeroBase
のデータとは別に(たとえば、パスワードなど)、新しいヒーローを作成する際により多くのデータを受け取りたいと簡単に決めることができ、今ではそれらの追加フィールドを入れるクラスが既にあります。
HeroPublic
データモデル¶
HeroPublic
モデルを確認しましょう。
これは、APIからヒーローを読み取る際にid
フィールドが必要であることを宣言するだけです。APIから読み取られるヒーローはデータベースから取得されるため、データベースには常にIDがあります。
# 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 👇
# Code above omitted 👆
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
# Code below omitted 👇
# Code above omitted 👆
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
# Code below omitted 👇
👀 ファイル全体のプレビュー
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
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
更新されたドキュメントUIを確認する¶
FastAPIコードは上記と同じです。Hero
、HeroCreate
、HeroPublic
を使用しています。しかし、ここでは継承を使用してよりスマートな方法でそれらを定義します。
そのため、すぐにドキュメントUIにジャンプして、更新されたデータがどのように表示されるかを確認できます。
ヒーローを作成するためのドキュメントUI¶
ヒーローを作成するための新しいUIを見てみましょう。
素晴らしい!ヒーローを作成するには、name
、secret_name
、オプションでage
を渡すだけでよくなりました。
id
は渡さなくなりました。
ヒーローレスポンスを使用したドキュメントUI¶
少し下にスクロールして、レスポンススキーマを確認できます。
id
は必須フィールドであり、赤いアスタリスク(*)が付いています。
ヒーロー読み取りのパスオペレーションのスキーマを確認すると、更新されたスキーマも表示されます。
継承とテーブルモデル¶
これらのモデルの継承がいかに強力であるかを見てきました。
これは非常に簡単な例であり、少し…つまらないように見えるかもしれません。😅
しかし、テーブルに10または20個のカラムがあり、すべてのデータモデルにその情報をすべて複製しなければならないと想像してみてください。そうすると、継承によってその情報重複を回避できることの有用性がより明らかになります。
さて、これはおそらく非常に柔軟性が高いように見えるため、いつ継承を使用するか、そして何のために使用するかがあまり明確ではありません。
役立つ経験則をいくつか紹介します。
データモデルからのみ継承する¶
データモデルからのみ継承し、テーブルモデルからは継承しないでください。
これにより混乱を回避でき、テーブルモデルから継承する必要はありません。
テーブルモデルから継承する必要があると感じた場合は、代わりに、データモデルのみであり、HeroBase
のようにそれらのフィールドすべてを持つ基本的なクラスを作成します。
そして、他のデータモデルとテーブルモデルの両方に対して、データモデルのみであるその基本的なクラスから継承します。
重複を回避する - シンプルさを保つ¶
「神秘的な方法で」異なる概念を分離するなど、あるモデルから別のモデルを継承する理由を深く考える必要があるように感じるかもしれません。
場合によっては、データの作成、読み取り、更新などのモデルのように、使用できる単純な分離があります。それが迅速かつ明らかであれば、それを利用します。💯
そうでない場合は、モデルを分離するための深い概念的な理由についてあまり心配する必要はありません。重複を回避し、十分にシンプルなコードを維持して、それを理解できるようにしましょう。
2つのモデル間に多くの重複があることがわかった場合は、基本モデルを使用してその重複の一部を回避できる可能性があります。
しかし、重複を回避するために、継承によるモデルの複雑なツリーを作成することになった場合、いくつかのフィールドを複製する方がシンプルになる可能性があり、それの方が理解しやすく、保守しやすくなります。
将来、推論、プログラミング、保守、リファクタリングが最も簡単な方法を選択しましょう。🤓
継承は、SQLModel と同様、他のものと同様に、生産性を向上させるためのツールに過ぎません。これは、それらの主要な目的の1つです。何かが生産性を向上させていない場合(例:重複が多すぎる、複雑すぎる)、変更してください。🚀
要約¶
SQLModel を使用して複数のモデルを宣言できます。
- 一部のモデルはデータモデルのみです。これらはPydantic モデルでもあります。
- また、一部のモデルは、
table = True
という設定を持つことで、(データモデルであることに加えて)テーブルモデルにもなります。これらもPydantic モデルとSQLAlchemy モデルになります。
テーブルモデルのみがデータベースにテーブルを作成します。
そのため、他のすべてのデータモデルを使用して、アプリケーションのデータのスキーマの検証、変換、フィルタリング、およびドキュメント化を行うことができます。✨
継承を使用して、情報とコードの重複を回避できます。😎
そして、これらのモデルをすべてFastAPI で直接使用できます。🚀