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

2025/02/24に公開
Python
 はじめに本記事を書こうと思ったきっかけですが、
生成AIを利用したサービス開発ではPythonが多く利用されている
それに伴い、バックエンド（API）の開発もPythonとなるケースをよく耳にする
じゃあ、簡単なAPIはPythonで実装できるようにしておいて損はないと思った
です。

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

 本記事を読んでいただくときの注意事項私のPythonスキルについてですが、
ソースが読める
Pythonに関してAPI開発を含め、多少業務で開発経験がある（メインはJavaとC#です）
です。

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

 目的以下を取り扱っていきます
Pythonの代表的なフレームワークについて
FastAPIの実装方法について
今後

 開発環境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/などのフォルダを作成しエントリー

※直接実装せず独立したライブラリを作成してしまうというパターンもありますが、その辺は臨機応変にしています。

 実装内容本記事では、「家計簿に記録した情報を参照する」ような仕組みを考えていこうと思います。
クライアントから検索条件を指定
指定した検索条件をもとにSQLを実行
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="タイプ１", money=1234567890)
+     dummy_2 = History(no=5678, date = datetime.now(), user_name="test", inout="支出", category="タイプ１", 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図に示している通り、支出履歴テーブルはユーザーテーブル/カテゴリーテーブルとリレーションがあります。その設定のうち、支出履歴テーブルとユーザーテーブルのリレーションの実装イメージは以下になります。
収支履歴テーブルとユーザーのリレーション
収支履歴テーブルのuser_id: str = Field(foreign_key="user.user_id")で外部キーを設定
収支履歴テーブルのuser: Optional[User] = Relationship(back_populates="inout_histories")で支出履歴テーブルからユーザーテーブルにリレーション
ユーザーテーブルの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))で検索条件を指定
!select(User, Category, InoutHistories)で指定しているテーブルの順序は取得結果の参照する際に重要になってくるため、覚えておいてください
最後に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のファイル名などセンスがないなど、読みにくさなど感じたかと思いますが、今後文章力も向上出来たらと思っています。以上です！
package	description
main.py	FastAPIのエントリーポイント
api/	エンドポイント（FastAPIのルーティング設定）
schemas/	APIのリクエスト/レスポンス
models/	ドメインモデル（業務ロジックのデータ定義など）
services/	業務処理
respositories/	DBなど外部サービスとの接続
.env	環境変数
requirements.txt	使用ライブラリ一覧（完成後に作成）
key	例	概説
from	20250101	取得開始日時を指定
to	20250131	取得終了日時を指定
階層	構成	ID	概説
1	取得結果	results	取得結果の一覧（収支履歴のオブジェクトをリストで持つ）
2	ID	id	収支履歴ID
2	日付	date	日付
2	ユーザ―名	user_name	ユーザー名
2	収支種別	type	収入 or 支出
2	カテゴリー名	category	食費/生活雑貨/・・・etc
2	金額	money	収支金額
はじめに

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

目的

開発環境

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

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

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

構成

実装内容

実装

Step.0 事前準備

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

Step.2 ルーターの作成

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

リクエスト

レスポンス

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

リクエスト

レスポンス

ルーターの再定義

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

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

3. 今後

Discussion
