はじめに
スプレッドシートで管理していたクライミングの記録をリレーショナルデータベースで管理することで、情報の一元化と分析のしやすさを垣間見ました。以下の記事です(以降、この記事を「前回」と書きます)。
一方で、DB Browser for SQLiteとCSVインポートでのデータ登録は苦行でしたので、その課題を解決するため、データベースを扱うWebアプリを作成してみます。
目標
クライミングの記録をブラウザから操作できるWebアプリを作成します。
適用技術
Pythonで手軽にWebアプリを作成するならStreamlitという優れたフレームワークがあるのですが、少し手の込んだことをしようとすると行き詰まりがちなので、バックエンドにFastAPI、フロントエンドにhtmxという組合せにします。PythonからSQLを扱うライブラリには、FastAPIの作者が開発したSQLModelを使ってみました。
- Python 3.12
- FastAPI
- htmx
- SQLModel
データベースは、前回の記事で作成したSQLiteのファイルをそのまま使います。
ディレクトリ構成
climbdata/
├── main.py # アプリ本体(サーバー処理)
├── models.py # データベース定義
├── climb.db # 前回のSQLiteファイル(なければ自動生成)
└── templates/ # HTMLファイル置き場
└── index.html # メイン画面
必要なライブラリ
pip install fastapi uvicorn sqlmodel jinja2 python-multipart
データベース定義の実装 (SQLModel)
前回の記事で設計したER図を基に、SQLModelを使ってPythonのクラスとして実装します。
従来、「SQLAlchemyでデータベース用の定義」し、「Pydanticでデータ検証・API用の定義」をするという二度手間が発生していましたが、SQLModelでは1つのクラス定義でこれら2つの役割を兼ねることができます。
from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Relationship
# 岩場 (Area)
class Area(SQLModel, table=True):
__tablename__ = "areas"
id: Optional[int] = Field(default=None, primary_key=True)
name: str
rock_type: Optional[str] = None
location: Optional[str] = None
# ルート (ClimbRoute)
class ClimbRoute(SQLModel, table=True):
__tablename__ = "climb_routes"
id: Optional[int] = Field(default=None, primary_key=True)
area_id: int = Field(foreign_key="areas.id")
name: str
grade: Optional[str] = None
angle: Optional[int] = None
type: Optional[str] = None
feature: Optional[str] = None
# 訪問記録 (Visit)
class Visit(SQLModel, table=True):
__tablename__ = "visits"
id: Optional[int] = Field(default=None, primary_key=True)
area_id: int = Field(foreign_key="areas.id")
date: str
condition: Optional[str] = None
memo: Optional[str] = None
# トライ記録 (Attempt)
class Attempt(SQLModel, table=True):
__tablename__ = "attempts"
id: Optional[int] = Field(default=None, primary_key=True)
route_id: int = Field(foreign_key="climb_routes.id")
visit_id: int = Field(foreign_key="visits.id")
progress_pct: Optional[int] = None
result: Optional[str] = None
comment: Optional[str] = None
# DB接続設定
sqlite_file_name = "climb.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
engine = create_engine(sqlite_url)
def create_db_and_tables():
SQLModel.metadata.create_all(engine)
アプリ本体の実装 (main.py)
次に、FastAPIを使ってサーバー側の処理を作ります。まずは、Webアプリで実現したい機能を整理します。
- どの岩場で
- どのような時期に
- どのような特徴のルートを
- どれだけ登ったのか
これらを一覧表示できるように実装を進めます。
まずはデータ取得だけしてみる
いきなりWeb画面を作る前に、まずはSQLiteデータベースからデータを取得して、文字列(JSON)としてブラウザに表示するシンプルな処理を作ってみます。
以下の内容で main.py を作成します。/attempts_check というURLにアクセスすると、データベースの中身をそのまま返してくれるシンプルな機能です。
from fastapi import FastAPI, Depends
from sqlmodel import Session, select
from models import engine, create_db_and_tables, Area, ClimbRoute, Visit, Attempt
app = FastAPI()
# データベースセッションの管理
# リクエストのたびにDBを開き、処理が終わったら確実に閉じる
def get_session():
with Session(engine) as session:
yield session
@app.on_event("startup")
def on_startup():
create_db_and_tables()
@app.get("/attempts_check")
def read_attempts_check(db: Session = Depends(get_session)):
# SQLを書かずにPythonのメソッドチェーンで検索条件を書く(SQLModelの真骨頂)
statement = select(Attempt).order_by(Attempt.id.desc()).limit(10)
attempts = db.exec(statement).all()
# FastAPIは、リストやオブジェクトを自動的にJSON(文字列)に変換して返す
return attempts
ターミナルで以下を実行し、サーバーを立ち上げます。
uvicorn main:app --reload
ブラウザで http://127.0.0.1:8000/attempts_check にアクセスします。画面に[{"id": 1, "result": "RP"...}, ...] のような文字データが表示されれば成功です。
SQLModelでは、SQLのSELECT操作を、以下のようなPythonらしいメソッドチェーンで表現できます。
attempts = db.exec(select(Attempt).order_by(Attempt.id.desc()).limit(10)).all()
この一行は、裏側で以下のようなSQL文を発行し、さらに結果をPythonのオブジェクトに変換してくれます。
-
select(Attempt):Attemptテーブルからデータを選ぶ(SELECT * FROM attempts) -
.order_by(Attempt.id.desc()): IDが新しい順に並び替える(ORDER BY attempt_id DESC) -
.limit(10): 最新の10件を取得 -
db.exec(...): ステートメントをデータベースで実行する -
.all(): 結果をすべてリストとして受け取る
従来の書き方であれば文字列でSQLを組み立てる必要がありましたが、SQLModelなら、メソッドをつなげていくだけ(メソッドチェーン)で、直感的に書くことができます。
複数のテーブルと情報を結合する
情報の結合(SQLの JOIN にあたる操作)も、SQLModelならPythonコードで完結します。select の引数に複数のモデル(クラス)を指定し、.join() メソッドで結合条件を書きます。
先程の main.py に、以下のルーターを追加します。
@app.get("/attempts_check_join")
def read_attempts_check(db: Session = Depends(get_session)):
# 1. 取得したいモデルを select にすべて列挙する
# 2. join でテーブル同士をどう繋ぐか指定する
statement = (
select(Attempt, ClimbRoute, Area)
.join(ClimbRoute, Attempt.route_id == ClimbRoute.id) # Attempt と ClimbRoute を結合
.join(Area, ClimbRoute.area_id == Area.id) # ClimbRoute と Area を結合
.order_by(Attempt.id.desc())
.limit(10)
)
# results は (Attempt, ClimbRoute, Area) のタプルのリスト
results = db.exec(statement).all()
# タプルを崩して辞書に詰め直す
data = []
for attempt, route, area in results:
data.append({
"id": attempt.id,
"result": attempt.result,
"route_name": route.name, # ルート名
"grade": route.grade, # グレード
"area_name": area.name, # 岩場名
"rock_type": area.rock_type # 岩質
})
return data
上書き保存すると自動的にFastAPIサーバーが変更を反映して再起動してくれます。ブラウザから http://127.0.0.1:8000/attempts_check_join にアクセスすると、ルート名、グレード、岩場名、岩質が結合されて表示されます。
この結合処理は以下の手順で行われています。
-
select(Attempt, ClimbRoute, Area): カンマ区切りで複数のモデルを指定することで、指定したテーブルのデータを全部取ってくる -
.join(ClimbRoute, ...): 「どの項目で紐付けるか」を指定する。例えば、Attempt.route_id == ClimbRoute.idは、トライ記録が参照しているルートIDと、ルートテーブルのIDを一致させる意味 -
戻り値: クエリの結果
resultsには、3つのオブジェクトがセットになったタプルが入ってきます。これをfor attempt, route, area in results:で回すことで、それぞれのテーブル情報へ簡単にアクセスする
この要領で、複数のテーブルから自在に情報を取り出せそうです。
Webアプリ用の動作にする
データの取得が確認できたので、いよいよ、WebアプリらしくHTMLテンプレート(画面)に表示させる準備をします。表示に加えて、新しい記録を保存するための処理と、既存の記録を削除するための処理も加えましょう。
main.py は以下のようになります。
from fastapi import FastAPI, Request, Form, Depends
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session as DBSession, select
from models import engine, create_db_and_tables, Area, ClimbRoute, Visit, Attempt
app = FastAPI()
# HTMLテンプレートの場所を指定
templates = Jinja2Templates(directory="templates")
# DBセッション管理
def get_session():
with DBSession(engine) as session:
yield session
# 起動時にテーブル作成
@app.on_event("startup")
def on_startup():
create_db_and_tables()
# --- メイン画面の表示 ---
@app.get("/", response_class=HTMLResponse)
def read_root(request: Request, db: DBSession = Depends(get_session)):
# 登録フォーム用: エリア一覧
areas = db.exec(select(Area)).all()
# 履歴表示用: 4つのテーブルを結合して、IDではなく「名前」や「日付」を取得する
statement = (
select(Attempt, ClimbRoute, Area, Visit)
.join(ClimbRoute, Attempt.route_id == ClimbRoute.id)
.join(Area, ClimbRoute.area_id == Area.id)
.join(Visit, Attempt.visit_id == Visit.id)
.order_by(Attempt.id.desc())
.limit(10)
)
results = db.exec(statement).all()
# テンプレートで表示しやすいように辞書のリストに変換
attempts_data = []
for attempt, route, area, visit in results:
attempts_data.append({
"id": attempt.id,
"date": visit.date, # Visitテーブルの日付
"area_name": area.name, # Areaテーブルの名前
"route_name": route.name, # Routeテーブルの名前
"grade": route.grade,
"result": attempt.result,
"progress_pct": attempt.progress_pct
})
return templates.TemplateResponse("index.html", {
"request": request,
"areas": areas,
"attempts": attempts_data # 整形済みのデータ
})
# --- htmx用:動的プルダウン ---
# エリアIDを受け取り、そのエリアのルート選択肢を返す
@app.get("/routes_options/{area_id}", response_class=HTMLResponse)
def get_routes_options(area_id: int, db: DBSession = Depends(get_session)):
# crag_id -> area_id, Route -> ClimbRoute
routes = db.exec(select(ClimbRoute).where(ClimbRoute.area_id == area_id)).all()
options_html = '<option value="" disabled selected>ルートを選択</option>'
for route in routes:
options_html += f'<option value="{route.id}">{route.name} ({route.grade})</option>'
return options_html
@app.post("/add_attempt")
def add_attempt(
date: str = Form(...),
area_id: int = Form(...), # crag_id -> area_id
route_id: int = Form(...),
result: str = Form(...),
db: DBSession = Depends(get_session)
):
"""
クライミング記録を追加
"""
# Visit を探すか作る
statement = select(Visit).where(Visit.date == date, Visit.area_id == area_id)
visit_obj = db.exec(statement).first()
if not visit_obj:
visit_obj = Visit(date=date, area_id=area_id)
db.add(visit_obj)
db.commit()
db.refresh(visit_obj)
# Attempt を登録
pct = 100 if result in ["RP", "OS", "FL"] else 0
attempt = Attempt(
route_id=route_id,
visit_id=visit_obj.id, # session_id -> visit_id
result=result,
progress_pct=pct
)
db.add(attempt)
db.commit()
return HTMLResponse(content="<div class='p-2 bg-green-200'>✅ 記録しました!</div>")
@app.delete("/delete_attempt/{attempt_id}")
def delete_attempt(attempt_id: int, db: DBSession = Depends(get_session)):
"""
クライミング記録を削除
"""
# IDでデータを検索
attempt = db.get(Attempt, attempt_id)
# データがあれば削除
if attempt:
db.delete(attempt)
db.commit()
# htmx用に「空のHTML」を返す
return HTMLResponse(content="")
get_routes_options 関数 (htmx連携)
ここがWebアプリとしての使い勝手を左右するポイントです。通常のWebアプリでは、岩場を選んだ後にルート一覧を更新するにはJavaScriptを記述する必要があります。しかし、この関数では単純に「HTMLの <option> タグの文字列」を返しているだけです。これを受け取った画面側(htmx)が、自動的にプルダウンの中身を差し替えてくれます。
add_attempt 関数 (データの登録)
ここでは、「訪問日」と「トライ(個別の記録)」の関係性を保つため、該当する訪問日がまだなければvisitsテーブルに自動的にレコードを追加する、という処理を入れています。CSV管理のときは自分で考えながらIDを割り振って行を追加していましたが、プログラムに任せれば楽になります。
フロントエンドの実装 (templates/index.html)
最後に、ブラウザで表示する画面(HTML)を作ります。以下の2つの技術を組み合わせ、モダンなUIを実現します。
- htmx: HTMLの属性を書くだけで、非同期通信(Ajax)や部分的な画面更新を行う
- Pico.css: HTMLタグを書くだけで自動的にデザインが整うCSSフレームワーク
templates フォルダの中に index.html を作成し、以下のコードを記述します。
<!DOCTYPE html>
<html lang="ja">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Climbing Log</title>
<script src="https://unpkg.com/htmx.org@1.9.10"></script>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@picocss/pico@2/css/pico.min.css">
<style>
:root {
/* 文字サイズと基本の余白を少し小さく */
--font-size: 15px;
--spacing: 0.75rem;
--form-element-spacing-vertical: 0.5rem;
--form-element-spacing-horizontal: 0.75rem;
}
/* コンテナ幅を少し狭めて間延びを防ぐ */
.container { max-width: 900px; }
/* テーブルのセルを詰める */
td, th { padding: 0.5rem 0.75rem; font-size: 0.9rem; }
</style>
</head>
<body>
<main class="container">
<h1>Climbing Log</h1>
<article>
<header>新規トライ記録</header>
<form hx-post="/add_attempt" hx-swap="outerHTML">
<div class="grid">
<label>
日付
<input type="date" name="date" required>
</label>
<label>
エリア
<select name="area_id"
onchange="htmx.ajax('GET', '/routes_options/' + this.value, '#route-select')"
required>
<option value="" disabled selected>エリアを選択</option>
{% for area in areas %}
<option value="{{ area.id }}">{{ area.name }}</option>
{% endfor %}
</select>
</label>
</div>
<label>
ルート
<select name="route_id" id="route-select" required>
<option value="" disabled selected>↑先に岩場を選んでください</option>
</select>
</label>
<div class="grid">
<label>
結果
<select name="result">
<option value="RP">RP (レッドポイント)</option>
<option value="OS">OS (オンサイト)</option>
<option value="FL">FL (フラッシュ)</option>
<option value="Fail">Fail (失敗)</option>
</select>
</label>
<button type="submit">記録する</button>
</div>
</form>
</article>
<h3>最近の記録</h3>
<figure>
<table>
<thead>
<tr>
<th>日付</th>
<th>エリア</th>
<th>ルート</th>
<th>結果</th>
<th>操作</th> </tr>
</thead>
<tbody>
{% for attempt in attempts %}
<tr>
<td>{{ attempt.date }}</td>
<td>{{ attempt.area_name }}</td>
<td>{{ attempt.route_name }}</td>
<td>{{ attempt.result }}</td>
<td>
<button
class="outline secondary"
style="padding: 0.2rem 0.5rem; font-size: 0.8rem;"
hx-delete="/delete_attempt/{{ attempt.id }}"
hx-confirm="本当に削除しますか?"
hx-target="closest tr"
hx-swap="outerHTML">
削除
</button>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</figure>
</main>
</body>
</html>
ブラウザから http://127.0.0.1:8000/ にアクセスすると、Webアプリっぽい画面場表示され、SQLiteデータベースの内容に対して記録や削除の処理ができるようになります。
htmxとPico.cssは、<head>タグ内で読み込むだけで設定が完了します。簡単です。htmxの恩恵を最も感じられるのが、岩場を選んだ瞬間にルートの選択肢が切り替わる部分です。
-
hx-trigger="change": プルダウンの値が変わったタイミングで発火する -
hx-get="/routes_options/...": 指定したURLへGETリクエストを送る -
hx-target="#route-select": 返ってきたHTML(<option>タグのリスト)で、IDがroute-selectの要素の中身を書き換える
JSONではなくHTMLを返すのがhtmxの哲学であり、バックエンドのロジックだけでUIの動きを完結させることができます。
また、Pico.cssを採用したことで、CSSファイル作成なし、クラス名の追加もほぼなしで、それっぽいUIになりました。main タグに container クラスを一つ当てた以外は、標準的なHTMLタグを書いているだけです。
まとめ
SQLiteデータベースにWebアプリというユーザーインターフェイスを加えることで、情報の一元化とデータ分析ができる仕組みのプロトタイプができあがりました。
- SQLModel のおかげで、直感的なオブジェクト操作でDBを扱えるようになり、
- htmx のおかげで、複雑なJavaScriptフレームワークを使わずにモダンなUIになり、
- FastAPI のおかげで、これらを手軽に構築できました。
今回のWebアプリはシンプルなものですが、同じ考え方で、様々な対象に応用できるのではと思います。
