tabi-bit · bobtaroh · Aug 1, 2026
@@ -51,6 +51,25 @@ server/
 - HTTPメソッド適切な使用（GET, POST, PUT, DELETE, PATCH）
 - レスポンス形式は一貫性を保つ
 
+#### 依存/パラメータの書き方（Annotated 必須）
+
+- FastAPI の `Depends` / `Query` / `Path` / `Header` / `Body` は **必ず `Annotated` 形式**で書く。
+  デフォルト値へ直接代入する旧形式は禁止（`ruff` の `FAST002` で CI が自動検出する）。
+
+  ```python
+  # NG（旧形式）
+  async def h(db: AsyncSession = Depends(get_db_session)): ...
+
+  # OK（Annotated）
+  async def h(db: Annotated[AsyncSession, Depends(get_db_session)]): ...
+  ```
+
+- **DI（`Depends`）系引数は関数シグネチャの先頭に置く**。Annotated 形式はデフォルト値を
+  持たないため引数順の制約がなく、依存を先頭・パスパラメータ/ボディを後ろに並べられる。
+- デフォルト値がある場合は Annotated の外に出す: `q: Annotated[int, Query(ge=0)] = 0`。
+- 検証は `server/scripts/verify.sh` を実行する（ruff check(FAST002) / ruff format / mypy / import+OpenAPI 生成の
+  全項目がゲート）。リファクタ前後で走らせて回帰がないことを確認する。CI (`ci_server.yml`) でも同スクリプトを実行する。
+
 ### 2. データベース操作
 
 - SQLAlchemyのORM使用

@@ -90,6 +90,10 @@ jobs:
         working-directory: ./server
         run: uv sync
 
+      - name: Lint / type-check / smoke (ruff + mypy + FastAPI import)
+        working-directory: ./server
+        run: ./scripts/verify.sh
+
       - name: Run Alembic migrations
         working-directory: ./server
         run: dotenvx run --env-file ../.env -- uv run alembic -n testdb upgrade head

@@ -62,10 +62,7 @@ async def create_block(db: AsyncSession, block: BlockCreate, page_id: int) -> Bl
     # location等のリレーションを事前取得(Eager Load)して返し直す
     stmt = (
         select(Block)
-        .options(
-            selectinload(Block.location),
-            selectinload(Block.destination_location)
-        )
+        .options(selectinload(Block.location), selectinload(Block.destination_location))
         .where(Block.id == db_block.id)
     )
     result = await db.execute(stmt)
@@ -84,9 +81,7 @@ async def find_blocks(db: AsyncSession, page_id: int) -> list[Block]:
     Returns:
         list[Block]: ブロックリスト
     """
-    result = await db.execute(
-        _block_with_relations().where(Block.page_id == page_id)
-    )
+    result = await db.execute(_block_with_relations().where(Block.page_id == page_id))
     return list(result.scalars().all())
 
 

@@ -138,9 +138,7 @@ async def delete_sent_notifications_for_block(
     return result.rowcount or 0
 
 
-async def delete_sent_notifications_for_page(
-    db: AsyncSession, *, page_id: int
-) -> int:
+async def delete_sent_notifications_for_page(db: AsyncSession, *, page_id: int) -> int:
     """指定 page 配下 block の送信済み予約を全て削除する。
 
     Page.date が変更されると配下の全 block の絶対日時が動くため、まとめて再通知させる。
@@ -272,7 +270,9 @@ async def list_notification_candidates(db: AsyncSession) -> list[NotificationCan
 
     filtered: list[dict] = []
     for r in rows:
-        absolute = _compose_absolute_start(r["page_date"], r["start_time"], r["timezone"])
+        absolute = _compose_absolute_start(
+            r["page_date"], r["start_time"], r["timezone"]
+        )
         if absolute is None:
             continue
         threshold = now_utc + timedelta(minutes=r["minutes_before"])
@@ -284,11 +284,11 @@ async def list_notification_candidates(db: AsyncSession) -> list[NotificationCan
         return []
 
     location_ids: set[int] = set()
-    for r in filtered:
-        if r["location_id"] is not None:
-            location_ids.add(r["location_id"])
-        if r["destination_location_id"] is not None:
-            location_ids.add(r["destination_location_id"])
+    for item in filtered:
+        if item["location_id"] is not None:
+            location_ids.add(item["location_id"])
+        if item["destination_location_id"] is not None:
+            location_ids.add(item["destination_location_id"])
 
     location_name_map: dict[int, str] = {}
     if location_ids:

@@ -45,9 +45,7 @@ async def find_pages(db: AsyncSession, trip_id: int) -> list[Page]:
     Returns:
         list[Page]: ページリスト
     """
-    result = await db.execute(
-        _page_with_relations().where(Page.trip_id == trip_id)
-    )
+    result = await db.execute(_page_with_relations().where(Page.trip_id == trip_id))
     return list(result.scalars().all())
 
 

@@ -8,7 +8,9 @@
 
 def _trip_with_relations():
     return select(Trip).options(
-        selectinload(Trip.pages).selectinload(Page.blocks).options(
+        selectinload(Trip.pages)
+        .selectinload(Page.blocks)
+        .options(
             selectinload(Block.location),
             selectinload(Block.destination_location),
         )

@@ -15,7 +15,9 @@ def format_title(start_time: datetime, subscriber_timezone: str) -> str:
     return f"next {local.strftime('%H:%M')}"
 
 
-def _format_location_line(location_name: str | None, destination_name: str | None) -> str | None:
+def _format_location_line(
+    location_name: str | None, destination_name: str | None
+) -> str | None:
     """場所行 (body の 2 行目) を組み立て。両方 null なら None を返して省略する。
 
     現状 UI では destination_name は常に null (destination_location を設定する UI 未実装) だが、

@@ -35,7 +35,10 @@ def verify_cloud_scheduler_oidc(request: Request) -> None:
     settings = get_settings()
 
     if settings.notify_tick_dev_bypass_oidc and settings.environment == "development":
-        logger.warning("OIDC verification bypassed (dev flag): environment=%s", settings.environment)
+        logger.warning(
+            "OIDC verification bypassed (dev flag): environment=%s",
+            settings.environment,
+        )
         return
 
     if (

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from typing import Annotated
 
 from fastapi import APIRouter, Depends
@@ -43,7 +44,7 @@ async def get_blocks(
     _: Annotated[int, Depends(require_page_access)],
     db: Annotated[AsyncSession, Depends(get_db_session)],
     page_id: int,
-) -> list[Block]:
+) -> Sequence[Block]:
     """
     説明:
 

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from typing import Annotated
 
 from fastapi import APIRouter, Depends
@@ -44,7 +45,7 @@ async def get_pages(
     _: Annotated[int, Depends(require_trip_access)],
     db: Annotated[AsyncSession, Depends(get_db_session)],
     trip_id: int,
-) -> list[Page]:
+) -> Sequence[Page]:
     """
     説明:
 

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Request, Response
@@ -50,7 +51,7 @@ async def create_trip(
 async def list_trips(
     _: Annotated[None, Depends(require_basic_auth)],
     db: Annotated[AsyncSession, Depends(get_db_session)],
-) -> list[Trip]:
+) -> Sequence[Trip]:
     """
     説明:
 

@@ -10,12 +10,16 @@
 
 
 class LocationBase(BaseModel):
-    google_place_id: str | None = Field(default=None, max_length=GOOGLE_PLACE_ID_MAX_LENGTH)
+    google_place_id: str | None = Field(
+        default=None, max_length=GOOGLE_PLACE_ID_MAX_LENGTH
+    )
     name: str = Field(min_length=1, max_length=LOCATION_MAX_NAME_LENGTH)
     address: str | None = None
     latitude: float | None = None
     longitude: float | None = None
-    website_uri: AnyHttpUrl | None = Field(default=None, max_length=WEBSITE_URI_MAX_LENGTH)
+    website_uri: AnyHttpUrl | None = Field(
+        default=None, max_length=WEBSITE_URI_MAX_LENGTH
+    )
 
     @field_serializer("website_uri")
     def _serialize_website_uri(self, v: AnyHttpUrl | None) -> str | None:

@@ -26,9 +26,7 @@ def _check_walica_url(cls, value: str | None) -> str | None:
             return value
         parsed = urlparse(value)
         if parsed.scheme not in ("http", "https") or parsed.hostname != WALICA_HOSTNAME:
-            raise ValueError(
-                "walica_url must be an http(s) URL on walica.jp"
-            )
+            raise ValueError("walica_url must be an http(s) URL on walica.jp")
         return value
 
     @model_validator(mode="after")

@@ -28,6 +28,7 @@ dev = [
     "pytest>=8.4.1",
     "pytest-asyncio>=1.1.0",
     "ruff>=0.12.7",
+    "types-nanoid>=2.0.0.20260508",
     "types-passlib>=1.7.7.20250602",
 ]
 [tool.pytest.ini_options]
@@ -36,3 +37,15 @@ asyncio_default_fixture_loop_scope = "function"
 
 [tool.ruff]
 exclude = ["migrations/versions"]
+
+[tool.ruff.lint]
+# 既定 (E, F) に加え、FastAPI の依存/パラメータを Annotated 形式で書くことを強制する。
+# FAST002: `x: T = Depends(...)` / `x: T = Query(...)` を禁止し `x: Annotated[T, ...]` を必須化。
+# ※ FAST001 (冗長な response_model) / FAST003 は既存コードに多数該当するため今は有効化しない。
+extend-select = ["FAST002"]
+
+[tool.mypy]
+# firebase-admin は型スタブ・py.typed マーカーを配布していないため import-untyped を無視する。
+[[tool.mypy.overrides]]
+module = "firebase_admin.*"
+ignore_missing_imports = true
@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+#
+# リファクタリング用の検証ハーネス。
+# DB を起動せずに「壊れていないか」「スタイル規約を満たすか」を高速に確認する。
+# 全項目 (ruff check / ruff format / mypy / smoke) はゲート = 失敗したら非0で終了する。
+#
+# 実行内容:
+#   1. ruff check   … lint。FAST002 により非-Annotated な依存/パラメータを検出する
+#   2. ruff format  … フォーマット差分の有無 (--check)
+#   3. mypy         … 型チェック
+#   4. smoke        … ダミー DB URL でアプリを import し OpenAPI を生成できるか
+#                     (FastAPI は依存注入グラフとシグネチャを登録/スキーマ生成時に検証するため、
+#                      Annotated の誤りや引数順の破綻はここで必ず落ちる。DB 接続は不要)
+#
+# 使い方:  server/ ディレクトリで `./scripts/verify.sh`
+# 終了コード: いずれか失敗で非0 (CI にそのまま組み込める)
+
+set -uo pipefail
+cd "$(dirname "$0")/.." || exit 1
+
+fail=0
+
+# ゲート: ベースラインでクリーンな項目。失敗したら非0で終了する。
+gate() {
+  local label="$1"; shift
+  echo "=== [gate] ${label} ==="
+  if "$@"; then
+    echo "  ✓ ${label}"
+  else
+    echo "  ✗ ${label} FAILED"
+    fail=1
+  fi
+  echo ""
+}
+
+gate "ruff check (FAST002 で Annotated 強制)" uv run ruff check app/
+gate "ruff format --check"                    uv run ruff format --check app/
+gate "mypy"                                    uv run mypy app/
+
+echo "=== smoke: import + OpenAPI 生成 ==="
+if DATABASE_URL="postgresql://u:p@localhost:5432/dummy" \
+   NOTIFICATIONS_ENABLED=false \
+   COOKIE_SECRET_KEY=dummy \
+   API_DOCS_USERNAME=dummy \
+   API_DOCS_PASSWORD=dummy \
+   uv run python -c "from app.main import app; n=len(app.openapi()['paths']); assert n>0; print(f'  OpenAPI OK: {n} paths')"; then
+  echo "  ✓ smoke"
+else
+  echo "  ✗ smoke FAILED"
+  fail=1
+fi
+echo ""
+
+if [ "$fail" -eq 0 ]; then
+  echo "✅ すべての検証をパスしました"
+else
+  echo "❌ 検証に失敗した項目があります"
+fi
+exit "$fail"