LoginSignup
0
0

More than 1 year has passed since last update.

@sand(sand mypress)

FastAPI と SQL Databases(SQLAlchemy)

Last updated at Posted at 2022-08-25

この記事はFastAPIでのSQLAlchemy ORMの使用方法を解説した以下の公式サイトを読んだ備忘録です。ValidationのためにPydanticクラスとの併用になりますが、全体的に冗長性は否めません。

FastAPI 公式サイト - SQL (Relational) Databases

（注意）SQLModelについて
また他方、FastAPIの作者自らSQLModelを作成しており今後はそちらの方に移行していくものと予想されます。SQLModelは Pydantic と SQLAlchemyを統合したものとなっており、コードの冗長性を減らす方向となっています。
SQLModel 公式サイト
FastAPI SQLModel 入門 - Qiita 2022/08/30　追加

1.ソースコードと解説

ここではORM として SQLAlchemy ORMを用い、 SQL (Relational) DatabasesとしてSQLiteを利用します。ちなみにSQLAlchemy　はasync/await と互換性がないことに注意してください。
SQLAlchemy 1.4 Documentation

pip install sqlalchemy

バージョンを確認します。

>>> import sqlalchemy  
>>> sqlalchemy.__version__                        
'1.4.27'

ソースコードツリーです。

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

1-1. database.py

database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

sessionmaker

sessionmakerでSessionクラスのオブジェクトが作られますが、これによってORMがdatabase をハンドルできるようになります。

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Declarative Extensions

SQLAlchemy ORM のテーブルとクラスのマッピングはDeclarative Extensionsによって行われます。クラスはdeclarative base classの拡張として定義されることでテーブルとのマッピングを維持できるようになります。

Base = declarative_base()

1-2. models.py

SQLAlchemy ORMによるクラスの定義です。declarative base classの拡張として定義されることで自動的にテーブルとのマッピングが維持されます。

models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

relationship

以下のようにSQLAlchemy ORM の relationship を定義しています。
Userオブジェクト user の user.items は、 Itemのowner_idがuser.idに等しいItemオブジェクトの配列となります。
逆にitem.ownerによって、owner_idによって指し示されるUserオブジェクトにアクセスできるようになります。

    items = relationship("Item", back_populates="owner")
    ---
    owner = relationship("User", back_populates="items")

1-3. schemes.py

ここではPydanticクラスを定義します。（個人的な見解ですが、上で定義したSQLAlchemyクラスを考えれば冗長性は否めません。）

Pydantic はデータ検証のためのPythonライブラリです。Pythonのtype hintが使われます。ですからPydanticクラスはAPIの入り口であるパスオペレーション関数で使われることが一般的です。

APIのCreate/Readで使うクラス定義は小さな差異があることが一般的ですが、初めに両者の共通部分を定義します。

共通部分 Create Read
ItemBase ItemCreate Item
UserBase UserCreate User

例えばユーザ情報をCreateするときはpasswordが必要ですが、Readするときはpasswordを含む必要がありません。

schemas.py
from typing import List, Union
from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

1-4. crud.py

databaseとのインターフェースを司るユーティリティの集まりです。ここではCRUD(Create, Read, Update, Delete)のCreateとReadのみに絞っています。

crud.py
from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

ここで各関数のreturnがSQLAlchemy modelsまたは SQLAlchemy models のリストであり、Pydantic modelsでないことに注意してください。 これは次のmain.pyでもう一度指摘します。

1-5. main.py

main.py
from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    print(user)
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

まずは前に述べたSQLAlchemy ORM のdeclarative base classのMetaData.create_all() methodを使って、テーブルの作成を行います。

models.Base.metadata.create_all(bind=engine)

A dependency with yield and try

Dependencies with yield
get_db()はdependencyとしてpath operationsから呼ばれ、database sessionを生成し返します(yield)。最終的にAPIのレスポンスを返したときにfinally部分が実行されます。つまりdb.close()が実行されます。
ちなみにyieldを含むget_db()関数は、Pythonのジェネレータ関数です。

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

orm_mode

前に指摘したように、crud.pyの各関数のreturnがSQLAlchemy modelsまたは SQLAlchemy models のリストであり、Pydantic modelsでありません。しかしpath operationではresponse_model=schemas.Userのように、response_modelではPydantic modelsが指定されています。
これはorm_modeの指定により、SQLAlchemy modelsからPydantic modelsの要素が自動に抽出されていることを意味しています。この時通常のようにfiltering や validationが行われます。

今回は以上です。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0