Go to Qiita Advent Calendar 2024 Top
LoginSignup
19
15

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

@yslius

SQLAlchemyを用いたFastAPIチュートリアル

Last updated at Posted at 2021-12-23

内容

SQLAlchemyを用いたFastAPIの実装について、公式サイトを参考に作っていきます。

関連記事

SQLAlchemyを用いたFastAPIチュートリアル　パスワードハッシュ化編
SQLAlchemyを用いたFastAPIチュートリアル　Basic認証編
SQLAlchemyを用いたFastAPIチュートリアル　JWT認証編

※記事ができ次第リンク貼ります。

環境

Windows10
Python 3.7.6
fastapi 0.70.1
SQLAlchemy 1.4.28

環境構築

先ずは作業ディレクトリにて仮想環境を作る

>python -m venv venv

FastAPIとSQLAlchemyをインストールする

>pip install fastapi sqlalchemy

今回はwindows環境でFastAPIの動作確認をしたいため、Uvicornをインストールする

>pip install uvicorn

ファイル構成

公式サイトに倣って以下のファイル構成でFastAPIを構築する

└─ fastapi_app
    ├─ crud.py
    ├─ database.py
    ├─ main.py
    ├─ models.py
    └─ schemas.py

今回はWindows環境上で動かしているため、公式サイトのコードに多少手を加えてます。

1.データベースを扱う処理を作る

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./fastapi_app.db"

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

Base = declarative_base()

SQLAlchemyのcreate_engine関数でengineを作り、sessionmaker関数でengineを用いてSessionLocal変数を作っておきます。SessionLocalは後ほどmainファイルで使用します。
declarative_base関数を使ってBase変数を定義しておき、次のモデルで使用します。

2.モデルを作る

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")

先ほど作成したBaseを継承してクラスを定義します。ここで定義したクラスが名データベースのテーブル名、クラス変数がカラム名、Column関数を使った定義がカラムの定義になります。
Column関数内でデータの型、primary_key、indexを設定します。

また、relationship関数を用いてUserとItemに双方向の関係を持たせています。

3.Pydanticモデルを作る

Pydanticを用いてスキーマを作ります。
Pydanticについてはこちらの記事が分かりやすかったです。
https://qiita.com/0622okakyo/items/d1dcb896621907f9002b

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


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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

BaseModelを継承して、ItemBase、UserBaseモデルを作り、ItemCreate/Itemクラス、UserCreate/Userクラスを定義します。
ここで定義したクラスをmain内で使用していきます。

4.CRUDを作る

CRUDとは、Create、Read、Update、Deleteの略で、データベースを操作する処理を定義しておきます。

crud.py
from sqlalchemy.orm import Session
import models
import 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

Userを作る関数、Userを読み取る関数、Itemを作る関数、Itemを読み取る関数を定義しておきます。
パスワードは暗号化していなく平文のままですので、あくまで動作確認用としてお使いください。

5.mainアプリを作る。

最後の仕上げでmain内に今まで作成したものを反映し、APIを構築します。

main.py
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
import crud
import models
import schemas
from database import SessionLocal, engine
import uvicorn

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)):
    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


if __name__ == '__main__':
    uvicorn.run(app=app)

main内ではget_db関数を定義してして、databaseで定義したSessionLocalを用いてdb変数をyieldします。
yieldは少しわかりにくいですが、以下の記事が参考になります。
https://qiita.com/tomotaka_ito/items/35f3eb108f587022fa09

FastAPIのURL操作として、@app.getや@app.postを用いて、URLの定義、modelはschemaで定義したものを用います。
@app.getや@app.postの下に関数を定義し、引数は関数内で使用したい引数を付けます。URLに{user_id}と記載した場合は、引数でuser_idとして使用できます。
データベースを使用する場合は以下の引数を付けます。

db: Session = Depends(get_db)

さて、いよいよアプリを起動してみてFastAPIの動作を確認してみます。
mainの最後に以下のコードを記載してuvicornを使用して、FastAPIを立ち上げてみます。

if __name__ == '__main__':
    uvicorn.run(app=app)

main.pyを実行すると、以下のようなメッセージが出ると思います。

INFO:     Started server process [14692]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

http://127.0.0.1:8000/docs
docsにブラウザからアクセスして以下のページが開かれたら成功です！

image.png

FastAPIはこのdocsからGETたPOSTが簡単にできるため、非常に優れているといつも感じております。

19
15
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
19
15

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?