概要
サービスを作る上で、バックエンドを真面目に
FastAPIとRDSを用いて実装してみようと思い立ち、一通り実装した時に、
一番難しいと感じたテーブルリレーションの話についてまとめようと思います。
今回記事にした内容のプログラムは、
こちらにアップロードされているので、必要であれば参考にしてみてください。
FastAPI とSqlalchemy
FastAPI とSqlalchemy を用いたバックエンドサービス構成の実装自体は
この記事が最高にわかりやすく、多くのことを学ばせていただきました。
フォルダ構成を始め、どの部分を切り分けて構成すれば良いのかなど、この記事を追っていけばかなり理解が深まるはずです。
上の解説記事については、Sqlalchemy のAsync Session を用いた実装となっていたのですが、
私は自分で実装する際にAsync Session でリレーションが引っ張れないエラーに遭遇し、
その後Asyncでない従来のSessionで実装し直しました。
その結果リレーションは問題なくDBから引っ張れたのですが、Async Sessionでもエラーを解決し、
そちらでもいつでも構成できるようにチャレンジしてみたいなとは思っています。
FastAPI + Sqlalchemy の構成
全体としての実装は上に書いた通り、記事を参考にしていただきたいのですが、
私が今回フォーカスしたいと考えているのは、テーブル間のリレーションについてです。
上のFastAPI入門の記事に従うと、
-
api/models/にSqlalchemy を用いたテーブル定義 -
api/schemas/に pydantic を用いたCRUDリクエストの入力と出力の型定義 -
api/routers/にCRUDで実際にコールするエンドポイントのルーティング -
api/cruds/にCRUDのエンドポイントが呼ばれた時のロジック
を構築していくことになります。
この時、後述のテーブルのリレーションによって、
- models
- schemas
- cruds
の実装に変更が必要となります。
また、多対多の構成では、仮に後述の中間テーブルに対して付加情報が必要な時は、
実装に工夫が必要になります。
私が実装を行った時に少し詰まったことや、理解が難しかったところを中心にまとめることができればと思います。
テーブルリレーション
テーブル間のリレーションは、基本的に
- 独立している（互いに関係がない）
- １対１
- １対多
- 多対多
のどれかに該当します。
それぞれをSqlalchemy のModel やpydantic のスキーマ定義を用いて実装するか一通り理解していれば、
これからのバックエンドサービスの構築がとてもスムーズになるなと思いました。
私が今回実装したサービスは、この３つのタイプを全て使う必要があったため、
上のFastAPI入門の記事に加えて、自分で必要なリレーションを定義し実装しました。
これらのリレーションを実装する上でのポイントをまとめてみようと思います。
書いてみると少し長くなってしまったので、多対多の構成については記事を分けることにします。
１対１
ここでは、Model とScaler が１対１の関係を持っているとします。
つまり、Model はただ一つのScalerと紐ついてるという条件です。
api/models
それぞれのModel定義は以下のようになります。
back_populates を用いてリレーションをお互いに構築した後、
uselist=False とし、外部キーをunique なprimary_keyに対して設定することで、
１対１を実装することができます。
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from api.db import Base
class Model(Base):
__tablename__ = "model"
id = Column(Integer, primary_key=True, autoincrement=True)
scaler_id = Column(Integer, ForeignKey("scaler.id"))
scaler = relationship("Scaler", back_populates="model", uselist=False)
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from api.db import Base
class Scaler(Base):
__tablename__ = "scaler"
id = Column(Integer, primary_key=True, autoincrement=True)
model = relationship("Model", back_populates="scaler")
api/schemas
schemas は以下のように定義します。
ModelクラスはCRUD のR（Read)の時にリターンされる型ですが、
Model にはScaler が紐ついてリターンされることがある（Optionalにしている）
ため、scaler: Optional[ScalerBase] の1行が書かれていることがポイントです。
from pydantic import BaseModel
from typing import List, Optional
from api.schemas.scaler import ScalerBase
class ModelBase(BaseModel):
pass
class Config:
orm_mode = True
class Model(ModelBase):
id: int
scaler: Optional[ScalerBase]
class ScalerBase(BaseModel):
pass
class Config:
orm_mode = True
class Scaler(ScalerBase):
id: int
api/cruds
以下のように、model を取得する（CRUDのR）時は、
model に紐ついた scaler もまとめて引っ張るように、クエリに対して
.options(joinedload(model_model.Model.scaler))
を追加しているところがポイントです。
また、この関数による返り値は、
List[Tuple[model_model.Model]]
となっており、上で定義したModel オブジェクトが返されることがわかっているため、
FastAPIのSwaggerUIを用いることで、どのようなオブジェクトが返ってくるのかが一目瞭然になり、
動作確認が大変やりやすくなっています。
また、scaler を新規作成する時（CRUDのC）は、
紐つけたいmodelのIDを指定し、それに対応したmodelを引っ張ってきた上で、
model.scaler = new_scaler
のように指定し、model, new_scaler オブジェクトをどちらもコミットすることで、リレーションを持った形でテーブルに挿入できます。
async def get_models(db: Session) -> List[Tuple[model_model.Model]]:
models = db.query(model_model.Model).options(joinedload(model_model.Model.scaler))
return models.all()
async def create_scaler(
db: Session, scaler_create: scaler_schema.ScalerCreateRequest
) -> scaler_model.Scaler:
model = await model_cruds.get_model(db, scaler_create.model_id)
scaler = scaler_model.Scaler()
db.add(scaler)
db.commit()
model.scaler = scaler
db.add(model)
db.commit()
return scaler
１対多
１対１と比較的同様に、１対多の構成についても言及します。
たとえば、Project １つに対して、多数のModelが紐ついている構成とします。
この場合は、以下のように実装します。
api/models
以下のプログラムをみると、確かに、
Projectモデルは Modelと対応関係を持ち、複数所持していることがわかるように、modelsと複数形で書いています。
先ほどの１対１で記載されていたuselist=Falseが指定されていないため、
今回は複数で紐つくことが可能になります。
１対多の多に対応する Model オブジェクトについては外部キーと、親オブジェクトとなるproject が記載されています。
class Project(Base):
__tablename__ = "project"
id = Column(Integer, primary_key=True, autoincrement=True)
models = relationship("Model", back_populates="project")
class Model(Base):
__tablename__ = "model"
id = Column(Integer, primary_key=True, autoincrement=True)
project_id = Column(Integer, ForeignKey("project.id"))
project = relationship("Project", back_populates="models")
api/schemas
Project は複数のModelを所有することから、Listを指定しています。
from pydantic import BaseModel, Field
from typing import List, Optional
class ProjectBase(BaseModel):
id: int
class Config:
orm_mode = True
class Project(ProjectBase):
models: List[ModelBase]
api/cruds
model を作成する（CRUDのC)際は、
対応する Projectのidを指定してproject を引っ張った後に、
project.models.append(model)
としてコミットすることで、関係性を保存しています。
async def create_model(
db: Session, model_create: model_schema.ModelCreateRequest
) -> model_model.Model:
project = await project_cruds.get_project(db, model_create.project_id)
model = model_model.Model(
project_id = model_create.project_id,
)
db.add(model)
db.commit()
project.models.append(model)
db.add(project)
db.commit()
return model
まとめ
今回は、FastAPI + SQLAlchemy を用いてバックエンドAPIを作成する時の構成で、
テーブルリレーションによって気をつけることについてまとめました。
１対１、１対多のリレーションについて書きましたが、
一度自分で実装してみることが重要かと思います。
実際にサービスを作る時は、多対多の構成が必要になることも多いです。
多対多の構成も、今回の構成の応用編ということで、ここまでわかっていればあまり問題なく実装できるのですが、
はまりどころもあるため、別記事にまとめたいと思います。
FastAPIはSwaggerUIを自動で生成してくれるところなどが強力で、
きちんとRDSを用いてバックエンドの構成を行いたい時はFastAPIは良い選択肢だなと思ったので、
興味を持った方で触ったことがまだない人は、ぜひ自分で簡単なサービスを作ってみると面白いかなと思います。
今回はこの辺で。