🥑

FastAPIを利用してAPIサーバを作る3- APIエンドポイント実装 -

2023/02/16に公開

Python

API

FastAPI

tech

この記事ではFastAPIを利用してAPIサーバを作る2- プロジェクトの作成-で作成したプロジェクトフォルダを利用し、実際にプログラムの実装をしていきます。
今回は例として、jsonでリクエストされた商品番号から、商品情報を返す簡単なAPIを開発していきます。

実装手順

DBモデル作成(app/models/product.py)
CRUDファイル作成(app/cruds/product.py)
Routerファイル作成(app/routers/product.py)
mainファイル作成(app/main.py)
Schemaファイル作成(app/schemas/product.py)

[フォルダ構成]

.
├── app
│   ├── cruds
│   ├── logs
│   ├── models
│   ├── routers
│   ├── schemas
│   ├── util
│   ├── db.py
│   └── main.py
├── Dockerfile
├── docker-compose.yml
├── docker-compose.prod.yml
├── poetry.lock
├── Makefile
├── .env
└── pyproject.toml

実装

1. DBモデル作成(app/models/product.py)

product.pyファイルを作成
モデルファイルを作成。このファイルに記載するクラスを通して、DBとデータをやり取りしたり、データの登録・更新・削除などの処理を行う。

.
└─ app
   └── models
   　　 └ product.py

app/models/product.py

from sqlalchemy.ext.automap import automap_base
from db import engine

baseModel = automap_base()

class Product(baseModel):
    __tablename__ = 'product'

baseModel.prepare(engine, reflect=True)

pythonのO/Rマッパー(リレーショナルデータベースにおけるレコードとを対応付けを行う「O/Rマッピング」のためのフレームワークやライブラリ)としてsqlalchemyの「automap_base」を利用している。これにより、カラムなどののテーブル情報をDBの情報から自動マッピングしてくれる。
公式ドキュメントの記載を見るに、下記のメソッドでDBのメタデータ（カラムの定義やリレーションなどのテーブル情報）を取得してモデルを生成していると思われる。

baseModel.prepare(engine, reflect=True)

また、O/Rマッパーを提供している他フレームワークなどでも同様かと思いますが、基本的にDBテーブルに主キーが定義されている必要があります。
もし定義されていないと上記の記載だとエラーになるため、下記のようにmodelクラス内でレコードを一意に識別可能なキーを定義する必要があります。(複数のカラムを主キーとして指定し複合主キーと定義することも可能)

class Product(baseModel):
    __tablename__ = 'product'

    baseModel.prepare(engine, reflect=True)

    product_id = Column(Integer, primary_key=True)

2. CRUDファイル作成(app/cruds/product.py)

product.pyファイルを作成
このファイルに記載した関数を介してモデルを操作し、DBのデータを取得する。
今回はSQLAlchemyのsessionmakerの関数を使用することで生SQLを書かずにクエリを発行することにした。
こうすることで、SQLの可読性が上がり、外注エンジニア等が参画した際にわかりやすいことがメリットである一方で、
より処理効率を求めたい場合など、直接SQLで書いた方が良い場面もあり(bulk insertなど)、その場合は内容をコメントで書いてわかりやすくするなどして、直接SQLを書くことも可能とした。

.
└─ app
   └── cruds
   　　 └ product.py

app/cruds/product.py

from db import get_session
from models.Products import Products

# 商品番号から商品情報を取得
def getProductInfoByProductId(product_id):
    if session == None:
        session = get_session()

    product_info = session.query(Products).\
    filter(Products.product_id == product_id).\
    filter(Products.del_flg == 0).\
    first()
    
    return product_info

db.pyファイルを作成
このファイルに記載した関数を介してモデルを操作し、DBのデータを取得する。
今回はSQLAlchemyのsessionmakerの関数を使用することで生SQLを書かずにクエリを発行することにした。
こうすることで、SQLの可読性が上がり、外注エンジニア等が参画した際にわかりやすいことがメリットである一方で、
より処理効率を求めたい場合など、直接SQLで書いた方が良い場面もあり(bulk insertなど)、その場合は内容をコメントで書いてわかりやすくするなどして、直接SQLを書くことも可能とした。

.
└─ app
   └── db.py

app/db.py
このファイルにはDBとセッションを張るための設定を記載する。
crudsディレクトリ配下のファイルはこの「get_session()」を利用することにより、DBとセッションを貼り、各種CRUD操作を行います。

import os

import sqlalchemy
from sqlalchemy.orm import sessionmaker

DATABASE = "mysql"
USER = os.environ['DB_USER']
PASSWORD = os.environ['DB_PASSWORD']
HOST = os.environ['DB_HOST']
PORT = 3306
DB_NAME = os.environ['DB_NAME']

DATABASE_URL = '{}://{}:{}@{}:{}/{}?charset=utf8'.format(DATABASE, USER, PASSWORD, HOST, PORT, DB_NAME)

ECHO_LOG = False

engine = sqlalchemy.create_engine(DATABASE_URL, echo=ECHO_LOG, pool_pre_ping=True)

def get_session():
    SessionClass = sessionmaker(engine)
    return SessionClass()

osモジュールを利用することにより、環境変数から値を取得できます。
データベースの接続情報などはAWSのSystems Manager Parameter Storeを利用し、イメージの外で管理し渡すようにしました。
※開発環境でDockerを立ち上げる際の環境変数はルートディレクトリ配下の.envに記載します。

import os
HOST = os.environ['DB_HOST']

下記の部分でDBとの接続を実現するengineオブジェクトを生成しています。
なお、引数「echo」はtrueにすると、実行されたSQLを標準出力してくれるようです。
また引数「pool_pre_ping」をtrueにすることで、クエリを発行する前にDBとのコネクションを確認(コネクションが切れていれば再接続)してくれるようです。
この設定をしない場合、engineオブジェクトは標準でコネクションプールが有効なので、接続を使い回すようなっているため、再利用しようと思った接続が勝手に切れていた場合に Lost connection エラーになる事があるようです。

engine = sqlalchemy.create_engine(DATABASE_URL, echo=ECHO_LOG, pool_pre_ping=True)

3. Routerファイル作成(app/routers/product.py)

product.pyファイルを作成
このファイルにエンドポイントの定義と実際の処理を書いていく。ソースコードが肥大化しやすい部分のため、ある程度の可読性を持たせるためにDB総裁に関しては、先程作成したcrudクラスのメソッドを呼び出すようにしたり、汎用的に使えそうな関数はutilityクラスを作成するよう開発時は意識する。

.
└─ app
   └── routers
   　　 └ product.py

app/routers/product.py

from fastapi import APIRouter, status
from fastapi.responses import JSONResponse

import app.cruds.v1.product as productCruds

from schemas.v1.product import (
    GetProductRequest as GetProductRequestSchema,
    GetProductResponse as GetProductResponseSchema,
    )

router = APIRouter(
    prefix="/v1/product",
    tags=["商品情報API"]
)

@router.get("/get_product_info", response_model=GetProductResponseSchema)
def get_product_info(params: GetProductRequestSchema):
    product_info = productCruds.getProductInfoByProductId(params.product_id)
    
    if product_info:
        return GetProductResponseSchema(
            result="success",
            product_info=product_info
            )
    else:
        error_response = {
                "error": "InformationForTheSpecifiedProductINotFound.",
                "message": "指定された商品番号の情報が見つかりませんでした。"
            }
        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=error_response)

下記の部分でルーターオブジェクトを作成し、APIエンドポイントを定義していきます。
引数「prefix」でエンドポイントのURIで共通のプレフィックスを定義しています。今回の開発では、商品関連、注文関連など業務種別によりファイルを分けて管理することにしたため、「v1/product」とプレフィックスを定義し、商品に関するエンドポイントが記載されることを明示しています。
なお、「v1」を含めている背景としては、今後、外部公開しているAPIの改修が必要となった場合に、利用者に切替に伴う負担が少なくなるよう、平行利用期間を設けた上で従来API（v1）から新規API（v2）へ移行することを意図しています。
また、引数「tags」を記載することで、後で紹介するSwagger(API ドキュメンテーションツール)の表示にタグをつけてくれるため、APIの目的等がわかりやすくなります。

router = APIRouter(
    prefix="/v1/product",
    tags=["商品情報API"]
)

「@router.get」という関数デコレータの第一引数でプレフィックスに続くAPIエンドポイントの実際のURIを定義します。
また、引数「response_model」でレスポンスデータの型チェックをするためのクラス、続く「get_product_info()」の引数にもリクエストデータをチェックするためのクラス(GetProductRequestSchema)を指定します。
ここで指定するクラスは後ほどSchemaファイルを作成し定義します。

※関数デコレータ：関数を受け取り関数を返す関数。続いて記載する関数を受け取り、その実行前後にデコレータの関数の処理を実行するイメージ

@router.get("/get_product_info", response_model=GetProductResponseSchema)
def get_product_info(params: GetProductRequestSchema):

4. mainファイル作成(app/main.py)

main.pyファイルを作成
先程作成したルーターオブジェクトを作成しただけではエンドポイントが公開できず、FastAPIオブジェクトに読み込ませることで利用可能となる。
「app.include_router()」メソッドで読み込んでいるが、今回のプロジェクト構成の場合、目的によってルーターオブジェクトを分けているために、新しいルーターオブジェクトが作成されるたびに、この記載を追加していく必要がある。

.
└─ app
   └── main.py

app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from routers.v1 import product

app = FastAPI()

# cors対応 リクエスト元のドメインを登録
origins = [
    "https://www.hoge.jp",
    "http://localhost:3000",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"] 
)

app.include_router(product.router)

5. Schemaファイル作成(app/schemas/product.py)

.
└─ app
   └── schemas
   　　 └ product.py

product.pyファイルを作成
下記でリクエストデータおよびレスポンスデータの型チェックを行うクラスをそれぞれ定義していきます。
クラス変数に変数名とその型指定を記載していきます。

from pydantic import BaseModel, Field

class GetProductRequest(BaseModel):
    product_id: int

class GetProductResponse(BaseModel):
    result: str
    product_info: dict

例えば、「product_id」の型をbool型で定義して、str型をリクエストすると422(Unprocessable Entity)エラーで下記のようなエラーレスポンスbodyが返されます。
また、必須パラメータがリクエスト値に存在しない場合などもエラーが返されます。

{
  "detail": [
    {
      "loc": [
        "body",
        "product_id"
      ],
      "msg": "value could not be parsed to a boolean",
      "type": "type_error.bool"
    }
  ]
}