#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してみましょう!
# 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
# -*- 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行の接続先情報は適時書き換えてください。
# -*- 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というモデルを作成しました。
# 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操作ができれば、
今回の目的は達成です。お疲れ様でした!