🔰

Python/FastAPIとSQLModelによるAPIの構築

2025/02/24に公開

はじめに

本記事を書こうと思ったきっかけですが、

  • 生成AIを利用したサービス開発ではPythonが多く利用されている
  • それに伴い、バックエンド(API)の開発もPythonとなるケースをよく耳にする
  • じゃあ、簡単なAPIはPythonで実装できるようにしておいて損はないと思った

です。
(あとは、言語によるプレビュー数などの推移とか少し検証したいなとかも考えています。)

本記事を読んでいただくときの注意事項

私のPythonスキルについてですが、

  • ソースが読める
  • Pythonに関してAPI開発を含め、多少業務で開発経験がある(メインはJavaとC#です)

です。
そのため、間違った内容を含む可能性が多々あるので、参考程度にしていただければと思っています。

目的

以下を取り扱っていきます

  1. Pythonの代表的なフレームワークについて
  2. FastAPIの実装方法について
  3. 今後

開発環境

  • OS:Windows10 HOME
  • Python:Verion 3.12.1
  • DB:SQLite
  • エディタ:Visual Studio Code

本記事で取り扱わないこと

  • Pythonのインストール方法
  • Visual Studio Codeの環境設定
    上記についてわざわざ書かなくともいろいろわかりやすい記事もしくはYouTubeの動画など情報がありふれているのでそちらをご参照いただければと思います。

1. Pythonの代表的なフレームワークについて

ぱっとすぐに思いつくのは、「Django」「Flask」「FastAPI」の3つかなと思います。それぞれの特徴については以下の記事をご参照いただけるとよいかと思います。

https://engineer-life.dev/flask-django/

2. FastAPIの実装方法について

構成

フォルダの全体像は以下のようにしていこうかと思います。
書いてはいるものの、すべてのファイルを作成・実装しているわけではありません。

/sample_proj
    ├ main.py
    ├ api/
    ├ schemas/
    ├ models/
    ├ services/
    ├ repositories/
    ├ .env
    ├ requirements.txt

各ファイルもしくはフォルダの役割としては以下になります。

package description
main.py FastAPIのエントリーポイント
api/ エンドポイント(FastAPIのルーティング設定)
schemas/ APIのリクエスト/レスポンス
models/ ドメインモデル(業務ロジックのデータ定義など)
services/ 業務処理
respositories/ DBなど外部サービスとの接続
.env 環境変数
requirements.txt 使用ライブラリ一覧(完成後に作成)

ざっくりした作りにしていますが、「フォルダを分けていればよいよね?」というようなことではないと思いますので、基本的に上記のような構成で作成することがほとんどです。

上記では記載していませんが、私の場合、共通処理の実装時はcommons/などのフォルダを作成しエントリー
※直接実装せず独立したライブラリを作成してしまうというパターンもありますが、その辺は臨機応変にしています。

実装内容

本記事では、「家計簿に記録した情報を参照する」ような仕組みを考えていこうと思います。

  1. クライアントから検索条件を指定
  2. 指定した検索条件をもとにSQLを実行
  3. SQLの実行結果をクライアントに返却

    ※DB設計は得意ではないので一旦適当にやっています

実装

Step.0 事前準備

実装に入る前に最低限必要なライブラリをpip installしておきます。

# ①仮想環境の作成(実行後、ルートディレクトリにvenvというフォルダが作成されます)
python -m venv venv

# ②作成した仮想環境をアクティベート
.\venv\Scripts\activate

# ③最低限必要なライブラリをインストール
pip install fastapi uvicorn

①の手順で実施している仮想環境の作成については、任意実行でOKです。
ただ、同一のローカル環境でPythonの開発を実施したことがある場合、同じライブラリの違うバージョンが混在し、思わぬエラーにつながる可能性があると思い私は基本的に実施しています。

Step.1 エントリーポイントの作成

ルートディレクトリの直下にmain.pyを作成します。
※エントリーポイントに起動確認用として1つエンドポイントを定義していますが、基本的にエンドポイントはAPIRouterで作成し、エントリーポイントで追加していくというな構成が良いと思います。

main.py
from fastapi import FastAPI

app = FastAPI()

# 起動確認のための実装
@app.get("/")
def init_root():
    return {"message" : "Hello World !!"}

起動時のポートは指定をしている場合を除き、デフォルトの8000になります。
実際にアクセスしてみて、{"message" : "Hello World !!"}が表示されることを確認してみましょう。

Step.2 ルーターの作成

ルーターを独立させることで、エンドポイントをカテゴリーごとに分けることができるため、可読性があると考えています。今回はサンプルなので1カテゴリーで実装していきます。

実際に作成したいAPIのエンドポイントを定義していくために、api/histories.pyを作成し以下のように処理を実装します。

api/histories_router.py
from fastapi import APIRouter
router = APIRouter()

@router.get("/histories/get")
async def get_user(id: str):
    return {"user_id" : id}

※クエリパラメータでidを受け取り、そのままレスポンスで返却するような処理です。

次に、main.pyに対してルーターを追加します。

main.py
from fastapi import FastAPI
+ from api import users

app = FastAPI()

# 起動確認のための実装
@app.get("/")
def init_root():
    return {"message" : "Hello World !!"}

+ # APIRouterの追加
+ app.include_router(users.router)

省略しますが、この時点でlocalhost:8000/histories/get?id=1234でアクセスしてみると{"id" : 1234}と表示されるはずです。
※起動コマンドで--reloadを指定し、ホットリロードにしているため、サーバの再起動は不要です。

Step.3 リクエスト / レスポンスの検討

私は各APIで「APサーバに何を送り、APサーバから何を返却してほしいか」をおおよそ考えてからいます。

今回は上記のDBから収支履歴を取得するAPIを作成してみます。

リクエスト

リクエストではSQLを実行する際の検索条件を指定していきます。
検索条件としては、
1. 主キーで検索
2. それ以外で検索
があると思いますが、上記2で日付で検索できるような仕組みを考えていこうと思います。
※取得系なので、HTTPメソッドの基本的な使い方に順守しGETリクエストを想定しています。

リクエストパラメータは以下にします。

key 概説
from 20250101 取得開始日時を指定
to 20250131 取得終了日時を指定

レスポンス

レスポンスではSQLの実行結果を返却できるような構成を考えていきます。

階層 構成 ID 概説
1 取得結果 results 取得結果の一覧(収支履歴のオブジェクトをリストで持つ)
2 ID id 収支履歴ID
2 日付 date 日付
2 ユーザ―名 user_name ユーザー名
2 収支種別 type 収入 or 支出
2 カテゴリー名 category 食費/生活雑貨/・・・etc
2 金額 money 収支金額

Step4. リクエストとレスポンスのスキーマを作成

Step3で検討したデータモデルになるように以下のファイルを作成します。

  • schemas/histories_search_param.py
  • schemas/histories_results.py

リクエスト

単純に検索条件となるパラメータを変数としてSearchDateクラスに定義します。

schemas/histories_search_param.py
from pydantic import BaseModel

class SearchDate(BaseModel):
    from_dt: str
    to_dt: str

レスポンス

1階層目の「履歴」をHistoryクラス、2階層目の「取得結果」をHistoriesクラスに定義します。

schemas/histories_results.py
from datetime import datetime
from pydantic import BaseModel

class History(BaseModel):
    no: int
    date: datetime
    user_name: str
    inout: str
    category: str
    money: int

class Histories(BaseModel):
    results: list[History]

ルーターの再定義

Step2で作成したルーターを作成したリクエストとレスポンスで実行できるように修正します。
※実行確認用にダミーデータを返却するようにしているため、余分なソースコードも含まれています。

api/histories.py
+ from fastapi import APIRouter
+ from fastapi import APIRouter, Depends
+ from typing import List
+ from schemas.histories_search_param import SearchDate
+ from schemas.histories_results import History, Histories
+ from datetime import datetime

router = APIRouter()

+# ダミーデータを入れる用
+ def create_dummydasta():
+     dummy_1 = History(no=1234, date = datetime.now(), user_name="test", inout="収入", category="タイプ1", money=1234567890)
+     dummy_2 = History(no=5678, date = datetime.now(), user_name="test", inout="支出", category="タイプ1", money=1234567890)
+     return [ dummy_1, dummy_2 ]

- @router.get("/histories/get")
- async def get_user(id: str):
-     return {"user_id" : id}

+ @router.get("/histories/get", response_model=Histories)
+ async def get_histories(params: SearchDate = Depends()):
+     # debug用
+     results = Histories(results = create_dummydasta())
+     return results

実行結果としては、以下のようになります。

Step5. 業務処理/DBアクセス処理の作成

DBへのアクセスはSQLModelを利用するので以下のコマンドを事前に実行しておいてください。

pip install sqlmodel

SQLModelについてZennで以下の本があり、わかりやすかったです。(内容も多くなく、SQLModelについて記載されていたので読了にそこまで時間を要しませんでした。)
https://zenn.dev/mook_jp/books/sqlmodel-tutorial/viewer/intro

以下のファイルを作成します。各ファイルの役割は順に説明します。

  • models/user.py
  • models/category.py
  • models/inout_histories.py
  • repositories/db_config.py
  • repositories/histories_repositories.py
  • services/histories_services.py

modelsディレクトリ配下にDBのデータ項目を定義していきます。
※これが良いとは思いませんが、テーブルごとにファイルを分けています。

models/user.py
from sqlmodel import SQLModel,Relationship, Field
from typing import List, Optional
from datetime import datetime

# ユーザーテーブル
class User(SQLModel, table = True):
    user_id: str = Field(primary_key = True)
    user_name: str
    inout_histories: List["InoutHistories"] = Relationship(back_populates="user")
models/category.py
from sqlmodel import SQLModel,Relationship, Field
from typing import List, Optional
from datetime import datetime

# カテゴリーテーブル
class Category(SQLModel, table = True):
    category_id: str = Field(primary_key = True)
    category_name: str
    inout_histories: List["InoutHistories"] = Relationship(back_populates="category")
models/inout_histories.py
from sqlmodel import SQLModel,Relationship, Field
from typing import List, Optional
from datetime import datetime
from models.user import User
from models.category import Category

# 収支履歴テーブル
class InoutHistories(SQLModel, table = True):
    id: int = Field(default = None, primary_key = True)
    date: datetime
    user_id: str = Field(foreign_key="user.user_id")
    user: Optional[User] = Relationship(back_populates="inout_histories")
    inout: str
    category_id: str = Field(foreign_key="category.category_id")
    category: Optional[Category] = Relationship(back_populates="inout_histories")
    money: int

※主キーについては、見たままなので説明は省略します。
ER図に示している通り、支出履歴テーブルはユーザーテーブル/カテゴリーテーブルとリレーションがあります。その設定のうち、支出履歴テーブルとユーザーテーブルのリレーションの実装イメージは以下になります。

  • 収支履歴テーブルとユーザーのリレーション
    1. 収支履歴テーブルのuser_id: str = Field(foreign_key="user.user_id")で外部キーを設定
    2. 収支履歴テーブルのuser: Optional[User] = Relationship(back_populates="inout_histories")で支出履歴テーブルからユーザーテーブルにリレーション
    3. ユーザーテーブルのinout_histories: List["InoutHistories"] = Relationship(back_populates="user")でユーザーテーブルから収支履歴テーブルにリレーション

上記の2,3の設定がない場合、取得結果を参照する際にエラーになります。

次にrepositoriesディレクトリ配下にDBアクセス系の処理を作成していきます。

repositories/db_config.py
from sqlmodel import create_engine, Session, SQLModel

DATABASE_URL = "sqlite:///./inout.db"
engine = create_engine(DATABASE_URL, echo=True)

def create_db():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

db_config.pyでは、テーブル作成と、セッションの取得の処理を記載しました。
他のrepositoryを作成したとしても、共通処理として使えるように外だししている感じです。

repositories/histories_repositories.py
from sqlmodel import Session, select
from datetime import datetime
from models.user import User
from models.category import Category
from models.inout_histories import InoutHistories

def select_histories_date(session: Session, fromdt: datetime, todt: datetime):
    statement = (
        select(User, Category, InoutHistories)
        .join(User, User.user_id == InoutHistories.user_id)
        .join(Category, Category.category_id == InoutHistories.category_id)
        .where(
            (fromdt <= InoutHistories.date) & (InoutHistories.date <= todt)
        )
    )
    results = session.exec(statement).all()
    return results

histories_repositories.pyには、実際にDBから収支履歴を取得する処理を記載しています。
詳細にはわかりませんが、自分のイメージとしては、

  • select(User, Category, InoutHistories)では、select ○○○ from User, Category,InoutHistoriesというようなイメージ
  • join(User, User.user_id == InoutHistories.user_id)でユーザーIDでテーブル結合
  • where((fromdt <= InoutHistories.date) & (InoutHistories.date <= todt))で検索条件を指定

最後にservicesディレクトリ配下に業務処理を作成していきます。
今回は、特に凝った処理は考えていないのでリクエストで受け取ったstr型の日付をdatetime型に変換してレポジトリにデータを渡しているだけです。

services/histories_services.py
from fastapi import Depends
from sqlmodel import Session
from datetime import datetime
from repositories.histories_repositories import select_histories_date

def get_histroies(session: Session, from_dt: str, to_dt: str):
    # 日付データ変換
    datetime_from_dt = datetime.strptime(from_dt, f"%Y%m%d")
    datetime_to_dt = datetime.strptime(to_dt, f"%Y%m%d")
    # 開始・終了日を条件に検索しrouterに返却
    return select_histories_date(session, datetime_from_dt, datetime_to_dt)

Step5. ルーターとエントリーポイントを最終系へ

※デバッグ用に追加していたものは事前に削除しました。

api/histories.py
from fastapi import APIRouter
from fastapi import APIRouter, Depends
from schemas.histories_search_param import SearchDate
from schemas.histories_results import History, Histories
+ from repositories.db_config import get_session
+ from services.histories_services import get_histroies
from datetime import datetime
from typing import List

router = APIRouter()

@router.get("/histories/get", response_model=Histories)
async def get_histories(params: SearchDate = Depends()):
+     # debug用
+     results = Histories(results = create_dummydasta())
+     return results
+     results = get_histroies(session, from_dt = params.from_dt, to_dt = params.to_dt)
+     histories: List[History] = []
+     for user, category, inout_history in results:
+         histories.append(
+             History(
+                 id = inout_history.id,
+                 date = inout_history.date,
+                 user_name = user.user_name,
+                 inout = inout_history.inout,
+                 category = category.category_name,
+                 money = inout_history.money
+                 )
+             )
+     return Histories(results = histories)

ルーターで取得結果からレスポンスデータを作成するために、servicesの返却値をfor文で回しています。ここでfor user, category, inout_history in resultsとしていますが、定義するデータモデルの順番がrepositoriesで指定した順番になっています。

※起動確認用に追加していたものは事前に削除しています。

main.py
from fastapi import FastAPI
+ from repositories.db_config import create_db

+ # テーブルの作成
+ create_db()

app = FastAPI()

# APIRouterの追加
app.include_router(users.router)

エントリーポイントにデータモデルからテーブルを作成するための処理を追加しています。
初回起動時にテーブルが作成されます。(2回目以降の起動でも特にエラーは発生しないので、これでよいかなと思っていますが、間違っていたらすみません、、)

3. 今後

次回以降にPython関連を試してみる際、自分の中で以下3つがあり悩みどころです。

  • GCPにFastAPIをデプロイ
  • Gemini APIを利用した生成AI関連の実装
  • FastAPI×TypeScriptでのWebアプリケーション開発

3つ目が一番面白味がなさそうです。(多分、CORS設定をFastAPI側にするだけでとりあえずの実装はできそうな感じがするため)
順番的には、Gemini API → GCPへのデプロイでやっていこうかと思います。
今回作成したのももったいないので利用して、生成AIを組み込んだ収支管理アプリ(無駄はどこなのかの分析や、収支予測、収支からの貯金予測・・・etc)など作れたら面白そうと思っています。

ここまでお読みいただきありがとうございました。
FastAPIにあまり詳しくなかったり、Pythonのファイル名などセンスがないなど、読みにくさなど感じたかと思いますが、今後文章力も向上出来たらと思っています。以上です!

Discussion