LoginSignup
11
8

More than 1 year has passed since last update.

@hkyo(hkyo san)

FastAPI入門 〜環境構築からMySQL連携まで〜

Posted at

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で起動するポートの指定も出来ます。

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

表示結果

helloworld.png

無事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にアクセスします。

Screenshot from 2022-02-11 00-54-24.png

上記のような自動で生成されるswaggerUIが表示できればOKです!
Screenshot from 2022-02-11 01-00-17.png
swaggerUIからAPIをテストしてみましょう。上手くDBのCRUD操作ができれば、
今回の目的は達成です。お疲れ様でした!

11
8
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
11
8