Liquidsのロゴ Liquids

FastAPIでDBを扱う方法

FastAPI
SQLAlchemy

FastAPIでDBを扱うにはORMを使用します。
ORMとはObject Relational Mappingのことで、データベースをPythonのオブジェクトとして扱うことができます。

Pythonの代表的なORMにはSQLAlchemyがあります。

FastAPIの公式チュートリアルでもDBを扱うためにSQLAlchemyが使われているため、SQLAlchemyを使えば間違いないでしょう。

SQL (Relational) Databases - FastAPI

このWikiでもSQLAlchemyを使用してFastAPIでDBを扱う方法を紹介していきます。

また、PostgreSQLを題材にしますが、RDMBSの違いを吸収できるのがORMのメリットの1つなので、他のRDBMSでも同じようにして扱えるでしょう。

pip install sqlalchemy

また、PostgreSQLを扱うにはpsycopg2のインストールも必要です。

pip install psycopg2

テーブルを表すクラスを作成するにはdeclarative_base()により得られるクラスを継承します。

models.py
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Users(Base) ・・・

テーブル名は__tablename__で設定します。

models.py
class Users(Base): __tablename__ = 'users

カラムはColumnを用いて定義します。

models.py
from sqlalchemy import Column, String class Users(Base): __tablename__ = 'users' user_name = Column(String(), nullable=False)

NOT NULL制約をつけるにはColumnのオプションであるnullableFalseを設定します。

models.py
from sqlalchemy import Column, String class Users(Base): __tablename__ = 'users' user_name = Column(String(), nullable=False)

主キーを定義するにはColumnのオプションであるprimary_keyTrueを設定します。

models.py
from sqlalchemy import Column, String from sqlalchemy.dialects.postgresql import UUID class Users(Base): __tablename__ = 'users' user_id = Column(UUID(), nullable=False, primary_key=True) user_name = Column(String(), nullable=False)

また、複合主キーにする場合は複数のカラムにprimary_key=Trueを設定します。

外部キーの定義にはForeignKeyを使用します。
引数には文字列で他テーブルのカラムを<テーブル名.カラム名>の形式で指定します。

このコードではownerカラムにはusersテーブルのuser_idカラムに存在している値でなければ存在することができません。

models.py
from sqlalchemy import Column, String, ForeignKey from sqlalchemy.dialects.postgresql import UUID class Users(Base): __tablename__ = 'users' user_id = Column(UUID(), nullable=False, primary_key=True) user_name = Column(String(), nullable=False) class Post(Base): __tablename__ = 'post' owner = Column(UUID(), ForeignKey('users.user_id'), nullable=False, primary_key=True) post_id = Column(UUID(), nullable=False, primary_key=True)
models.py
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() from sqlalchemy import Column, String, ForeignKey from sqlalchemy.dialects.postgresql import UUID class Users(Base): __tablename__ = 'users' user_id = Column(UUID(), nullable=False, primary_key=True) user_name = Column(String(), nullable=False) class Post(Base): __tablename__ = 'post' owner = Column(UUID(), ForeignKey('users.user_id'), nullable=False, primary_key=True) post_id = Column(UUID(), nullable=False, primary_key=True)

テーブルの作成はmetadata.create_allで実施ができます。
テーブルの作成は最初に1度実行するだけでいいでしょう。

database.py
from sqlalchemy import create_engine engine = create_engine( # 適宜、自環境のURLに置き換える 'postgresql://postgres:pass@localhost:5432/example_db' )
create_tables.py
import models from database import engine models.Base.metadata.create_all(bind=engine)

作成するテーブルの定義はSTEP2を参照してください。

セッションメーカーはsessionmakerで作成ができます。
このセッションメーカーを利用してセッションを作成し、クエリを発行していくことになります。

database.py
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker # データベースURL engine = create_engine( 'postgresql://postgres:pass@localhost:5432/example_db' ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

セッションの取得自体はエンドポイント関数で行う必要はないが、FastAPIのドキュメントに沿った方法でセッションを取得してみる。

以下のように、STEP4で作成したSessionLocalからセッションを作成する。
Dependsについては ドキュメント を参照

main.py
from fastapi import FastAPI, Depends from database import SessionLocal app = FastAPI() def get_db(): db = SessionLocal() try: yield db finally: db.close() db = Depends(get_db) @app.post('/') def index(db=db): return None

クエリの発行はSTEP5で作成したdbから行う。

この例ではUsers(usersテーブル)に対して、user_idカラムが特定のIDであることを条件としてwhere句(filter)を使用しています。
また、first()は条件に合致した最初のレコードを取得するという意味です。
(全てのレコードならall())

main.py
from fastapi import FastAPI, Depends from database import SessionLocal from models import Users app = FastAPI() def get_db(): db = SessionLocal() try: yield db finally: db.close() db = Depends(get_db) @app.post('/') def index(db=db): user = db.query(Users).filter(Users.user_id == 'ab2a848c-6e3d-41b6-8315-2080d815f011').first() print(user.user_id) # ab2a848c-6e3d-41b6-8315-2080d815f011 print(user.user_name) # test user

INSERTやORDER BYなど他のSQLを実行する方法も、もちろんSQLAlchemyでは用意されています。

それらはこちらの記事が参考になります

models.py
from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() from sqlalchemy import Column, String, ForeignKey from sqlalchemy.dialects.postgresql import UUID class Users(Base): __tablename__ = 'users' user_id = Column(UUID(), nullable=False, primary_key=True) user_name = Column(String(), nullable=False) class Post(Base): __tablename__ = 'post' owner = Column(UUID(), ForeignKey('users.user_id'), nullable=False, primary_key=True) post_id = Column(UUID(), nullable=False, primary_key=True)
database.py
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker # データベースURL engine = create_engine( 'postgresql://postgres:pass@localhost:5432/example_db' ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
create_tables.py
import models from database import engine models.Base.metadata.create_all(bind=engine)
main.py
from fastapi import FastAPI, Depends from database import SessionLocal from models import Users app = FastAPI() def get_db(): db = SessionLocal() try: yield db finally: db.close() db = Depends(get_db) @app.post('/') def index(db=db): user = db.query(Users)\ .filter(Users.user_id == 'ab2a848c-6e3d-41b6-8315-2080d815f011')\ .first() print(user.user_id) # ab2a848c-6e3d-41b6-8315-2080d815f011 print(user.user_name) # test user