FastAPIとは

FastAPIとはAPI作成に特化したPythonのフレームワークです。
Python3.6以上で動作し、簡単にRESTfulなAPIを作成することができます。
また自動で生成されるswaggerUIドキュメントからAPIのテストなども行えます。

目次

環境構築
 HelloWorld
MySQL連携とCRUD操作

自身の環境

Ubuntu 20.04 LTS
Python 3.8.10
MySQL 8.0.28
SQLAlchemy 1.4.28

1. 環境構築

環境構築といってもFastAPIは非常に手軽で、pipでインストールするだけで簡単に使用することが出来ます。

install

pip install fastapi uvicorn

インストール出来ましたか？
今回はfastapi_sampleディレクトリを作成し、そちらにプログラムを置くことにします。

mkdir fastapi_sample
touch main.py

作成後のディレクトリの構造は以下の通りです。

ディレクトリ構造

fastapi_sample
└── main.py

2. helloworld

それではFastAPIでHelloWorldしてみましょう!

main.py

# FastAPIをインポート
from fastapi import FastAPI

# FastAPIのインスタンス作成
app = FastAPI()

# GETメソッドでルートURLにアクセスされたときの処理
@app.get("/")
async def root():
    return {"message": "Hello World"}

サーバー起動

uvicorn main:app --reload --port 8081

main : appの部分はファイル名 : インスタンス名です。
(今回はmain.pyというファイル名とプログラムの３行目作成したappというインスタンスを指定しています)
オプションとして--reloadと指定しています。こうすることでコード更新の度にサーバーを自動的に更新してくれるので便利です。
--portで起動するポートの指定も出来ます。

サーバーが起動したら早速ブラウザで確認してみましょう。

表示結果

無事JSONレスポンスが表示されていますね！

3. データベース連携

次はFastAPIからDBのCRUD操作ができるようにしてみましょう！
今回使用するDBはMySQLです。

まずはSQLAlchemyというORMをインストールします。

pip install sqlalchemy

次にORMの設定ファイルとモデル定義ファイルを作成します。

touch db_setting.py
touch db_model.py

作成後のディレクトリの構造は以下の通りです。

ディレクトリ構造

fastapi_sample
└── main.py
└── db_setting.py
└── db_model.py

db_setting.py

# -*- coding: utf-8 -*-

# DB操作用にsqlalchemyライブラリインポート
from sqlalchemy import create_engine
# DBの存在確認とDB作成を行うためにインポート
from sqlalchemy_utils import database_exists, create_database
# セッション定義用にインポート
from sqlalchemy.orm import sessionmaker, scoped_session

# モデル定義ファイルインポート
from db_model import Base

# 接続したいDBへの接続情報
user_name = 'NAME'
password = 'PASSWORD'
host = "HOST"
database_name = "fastapi_sample"

# バインディング
DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
    user_name,
    password,
    host,
    database_name,
)

# DBとの接続
ENGINE = create_engine(
    DATABASE,
    # 文字コード指定
    encoding="utf-8",
    #自動生成されたSQLを吐き出すようにする
    echo=True
)

# session変数にsessionmakerインスタンスを格納
session = scoped_session(
    # ORマッパーの設定。自動コミットと自動反映はオフにする
    sessionmaker(
        autocommit=False,
        autoflush=False,
        bind=ENGINE
    )
)
# DBが存在しなければ
if not database_exists(ENGINE.url):
    # DBを新規作成する
    create_database(ENGINE.url)

# 定義されているテーブルを一括作成
Base.metadata.create_all(bind=ENGINE)

# DB接続用のセッションクラス、インスタンスが作成されると接続する
Base.query = session.query_property()

ORMの設定ファイルです。14-17行の接続先情報は適時書き換えてください。

db_model.py

# -*- coding: utf-8 -*-

# sqlalchemyライブラリから使用する型などをインポート
from sqlalchemy import Column, Integer, String,DateTime
# CURRENT_TIMESTAMP関数を利用するためにインポート
from sqlalchemy.sql.functions import current_timestamp
# Baseクラス作成用にインポート
from sqlalchemy.ext.declarative import declarative_base

# Baseクラスを作成
Base = declarative_base()

# Baseクラスを継承したモデルを作成
# usersテーブルのモデルUsers
class Users(Base):
    __tablename__ = 'users'
    user_id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(20), nullable=False)
    mail = Column(String(50),nullable=False,unique=True)
    sex = Column(String(3),nullable=True)
    created_at = Column(DateTime, server_default=current_timestamp())

モデルの定義ファイルです。今回はUsersというモデルを作成しました。

main.py

# FastAPIインポート
from fastapi import FastAPI
# 型ヒントを行えるpydanticをインポート
from pydantic import BaseModel  

# 作成したモデル定義ファイルと設定ファイルをインポート
import db_model as m 
import db_setting as s 

# データクラス定義
# POSTとPUTで使うデータクラス
class UserBase(BaseModel):
    name : str
    mail : str
    sex : str

# FastAPIのインスタンス作成
app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}

# GETメソッドで /usersにアクセスしたときの処理
# ユーザーの全件取得
@app.get("/users", tags=["users"])
async def read_users():
    #DBからユーザ情報を取得
    result = s.session.query(m.Users).all()
    return result

# POSTメソッドで /usersにアクセスしたときの処理
# ユーザーの新規登録
@app.post("/users", tags=["users"])
async def create_user(data: UserBase):
    # Usersモデルを変数に格納
    user = m.Users()
    # セッションを新規作成
    session = s.session()
    s.session.add(user)
    try:
        #リクエストBodyで受け取ったデータを流し込む
        user.name = data.name
        user.mail = data.mail
        user.sex = data.sex
        #永続的にDBに反映
        session.commit()
    except:
        # DBへの反映は行わない
        session.rollback()
        raise
    finally:
        # 正常・異常どちらでもセッションは終わっておく
        session.close()

# DELETEメソッドで /usersにアクセスしたときの処理
# ユーザーの削除
@app.delete("/users/{id}", tags=["users"])
async def delete_user(id: int):
    # セッションを新規作成
    session = s.session()
    try:
        # 指定されたuser_idのユーザーを削除
        query = s.session.query(m.Users)
        query = query.filter(m.Users.user_id == id)
        query.delete()
        # 永続的にDBに反映
        session.commit()
    except:
        # DBへの反映は行わない
        session.rollback()
        raise
    finally:
        # 正常・異常どちらでもセッションは終わっておく
        session.close()

# PUTメソッドで /usersにアクセスしたときの処理
# ユーザーの更新
@app.put("/users/{id}", tags=["users"])
async def update_user(id: int, data:UserBase):
    # セッションを新規作成
    session = s.session()
    try:
        # ユーザー更新
        s.session.query(m.Users).\
        filter(m.Users.user_id == id).\
        update({"name" : data.name, "mail" : data.mail, "sex": data.sex})
        # 永続的にDBに反映
        session.commit()
    except:
        # DBへの反映は行わない
        session.rollback()
        raise
    finally:
        # 正常・異常どちらでもセッションは終わっておく
        session.close()

さきほど作成したmain.pyを更新します。
更新できたら以下のURLにアクセスします。

上記のような自動で生成されるswaggerUIが表示できればOKです!

swaggerUIからAPIをテストしてみましょう。上手くDBのCRUD操作ができれば、
今回の目的は達成です。お疲れ様でした!