Python + FastAPI マニュアル（VS Code / Windows 11 / Python初心者向け）

0. Python + FastAPI の目的と概要

FastAPI が向いていること

• スマホ/フロント向けの REST API（ログイン、CRUD、検索など）
• 社内ツールの 小～中規模 API（1台～数台で運用）
• Python の強み（データ処理/機械学習）を API 化して提供

このマニュアルのゴール

1. Windows 11 で Python / FastAPI の環境構築

1-1. まず入れるもの（最小）

1-2. Python インストール（Windows）

1. Python をインストール（公式インストーラ or Microsoft Store など）
2. PowerShell を開き、バージョン確認：

   python --version
   pip --version

1-3. 仮想環境（venv）を作る

# 作業フォルダへ移動（例）
cd C:\work

# プロジェクト作成
mkdir fastapi-sample
cd fastapi-sample

# venv 作成（.venv という名前にするのがおすすめ）
python -m venv .venv

1-4. venv を有効化（Activate）

# PowerShell
.\.venv\Scripts\Activate.ps1

# コマンドプロンプト（cmd）
.\.venv\Scripts\activate.bat

1-5. FastAPI / Uvicorn をインストール

python -m pip install --upgrade pip
pip install fastapi "uvicorn[standard]"

2. 標準的なフォルダ構成（迷わないための地図）

2-1. 最小構成（まず動かす）

fastapi-sample/
  .venv/
  main.py
  requirements.txt
  README.md

2-2. 実務寄りの標準構成（おすすめ）

fastapi-sample/
  .venv/
  app/
    __init__.py
    main.py              # FastAPI app の入口
    routers/             # ルーティング（Controller相当）
      health.py
      users.py
    schemas/             # リクエスト/レスポンス型（DTO相当）
      user.py
    services/            # 業務ロジック
      user_service.py
    db/                  # DB接続・Repository層
      session.py
      user_repo.py
    core/                # 設定・共通部品
      config.py
  tests/
    test_health.py
  requirements.txt
  .env.example
  .gitignore

3. VS Code に入れておくといい Extension

4. VS Code でのプロジェクト新規作成

4-1. フォルダを作って VS Code で開く

1. 例：C:\work\fastapi-sample を作成
2. VS Code → File → Open Folder... で開く
3. VS Code → Ctrl + Shift + @ でターミナルを開く

4-2. venv を作って有効化（VS Code 内ターミナル）

python -m venv .venv
.\.venv\Scripts\Activate.ps1

4-3. 依存を入れる

python -m pip install --upgrade pip
pip install fastapi "uvicorn[standard]"
pip freeze > requirements.txt

4-4. 入口ファイル（main.py）を作る

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
def health():
    return {"status": "ok"}

5. 開発サーバの起動・動作確認

5-1. Uvicorn で起動（リロード有効）

uvicorn main:app --reload --host 127.0.0.1 --port 8000

5-2. ブラウザで確認

• ヘルスチェック：http://127.0.0.1:8000/health
• APIドキュメント（Swagger UI）：http://127.0.0.1:8000/docs
• 代替UI（ReDoc）：http://127.0.0.1:8000/redoc

6. VS Code を使用した Debug 方法（ブレークポイント）

6-1. デバッグ設定（launch.json）を作る

1. 左の「実行とデバッグ」 → 「create a launch.json file」
2. 環境は Python を選ぶ
3. 以下の例のように uvicorn を module 起動する設定にする

例：.vscode/launch.json（最小）

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "FastAPI (uvicorn)",
      "type": "python",
      "request": "launch",
      "module": "uvicorn",
      "args": [
        "main:app",
        "--reload",
        "--host", "127.0.0.1",
        "--port", "8000"
      ],
      "jinja": false
    }
  ]
}

6-2. ブレークポイントで止める

1. main.py の return 行にブレークポイント
2. VS Code で「FastAPI (uvicorn)」を選んで ▶ を押す
3. ブラウザで /health を叩く
4. 止まったら、Variables / Watch / Call Stack で状態確認

7. リリース版の作り方と配布方法（実務で困らないために）

7-A. いちばんシンプル：ソース + requirements.txt を配布

1. 配布物を zip にする（例）：

   app/（または main.py）
   requirements.txt
   README.md
   .env.example

2. 配布先で venv 作成 → 依存インストール：

   python -m venv .venv
   .\.venv\Scripts\Activate.ps1
   pip install -r requirements.txt

3. 起動（本番は reload なし）：

   uvicorn main:app --host 0.0.0.0 --port 8000

7-B. Docker 配布（環境差分を消す）

（最小イメージ例：参考）

# Dockerfile（例）
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

7-C. Windows サービス化（常駐運用）

7-D. “exe に固める” はアリ？

7-1. 本番向けの最低チェックリスト

• 設定：.env を使うなら .env.example を用意し、機密は配布物に直書きしない
• ログ：標準出力だけでなくファイル/イベントログに落とす運用を決める
• ポート：FW 許可、競合回避、監視の設計
• 起動方法：手動/タスクスケジューラ/サービス化のどれにするか決める

2. 標準的なフォルダ構成（FastAPI版）

2-1. FastAPI プロジェクトフォルダ構成（代表）

2-2. 初心者が最初に触る場所（FastAPI）

3. Python の変数・型と基本文法

3-1. Python の主な型

3-2. if / switch / while / for / foreach

if（条件分岐）

if x > 10:
    print("large")
elif x > 5:
    print("middle")
else:
    print("small")

switch 相当（match：Python 3.10+）

match status:
    case 200:
        print("OK")
    case 404:
        print("Not Found")
    case _:
        print("Other")

while

i = 0
while i < 3:
    print(i)
    i += 1

for

for i in range(3):
    print(i)

foreach（Pythonでは for）

users = ["Alice", "Bob", "Carol"]

for user in users:
    print(user)

6. Python の変数定義とスコープ（外部変数 / クラス変数 / ローカル変数 / 定数）

6-1. Python における「変数定義」とは

x = 10
name = "Alice"
is_active = True

上記のように、= で値を代入した時点で変数が作られます。
型は 値の型 によって自動的に決まります。

6-2. スコープの種類（LEGBルール）

6-3. ローカル変数（Local）

def sample():
    x = 10  # ローカル変数
    print(x)

sample()
# print(x)  # ← エラー（関数の外からは見えない）

• 関数の中で定義された変数
• 関数の外からは参照できない
• FastAPI では「1リクエスト内の一時変数」になる

6-4. 外部変数（グローバル変数 / Global）

count = 0  # 外部（グローバル）変数

def increment():
    global count
    count += 1

6-5. クラス変数（Class Variable）

class User:
    role = "guest"  # クラス変数

    def __init__(self, name):
        self.name = name  # インスタンス変数

• クラス直下で定義される
• 全インスタンスで共有される
• 設定値・共通定数向き

6-6. インスタンス変数（参考）

u = User("Alice")
print(u.name)  # インスタンス変数
print(User.role)  # クラス変数

6-7. 定数の書き方（Python流）

API_TIMEOUT = 30
DEFAULT_ROLE = "guest"
MAX_RETRY_COUNT = 3

• 全て大文字＋アンダースコア
• 書き換えない前提の値
• 設定ファイル・定数モジュールにまとめる

6-8. まとめ（事故らないための指針）

• 基本は「ローカル変数＋引数＋戻り値」
• 外部（グローバル）変数は読み取り専用
• 共有したい値はクラス変数 or 設定クラス
• 定数は 大文字名＋書き換えない

12. 文字型・数値型の相互変換（Python）

12-1. 文字列 → 数値型

int への変換

value = "123"
num = int(value)

print(num)        # 123
print(type(num))  # <class 'int'>

float への変換

value = "3.14"
num = float(value)

print(num)        # 3.14
print(type(num))  # <class 'float'>

value = "abc"
num = int(value)  # ValueError

12-2. 数値型 → 文字列

count = 10
text = str(count)

print(text)       # "10"
print(type(text)) # <class 'str'>

文字列連結時の注意

# NG
# print("count=" + count)  # TypeError

# OK
print("count=" + str(count))
print(f"count={count}")

12-3. float → int（切り捨て・丸め）

切り捨て（truncate）

value = 3.99
num = int(value)

print(num)  # 3

四捨五入

value = 3.5
num = round(value)

print(num)  # 4

12-4. 文字列 → 数値（安全に変換する）

def to_int(value: str, default: int = 0) -> int:
    try:
        return int(value)
    except ValueError:
        return default

print(to_int("10"))     # 10
print(to_int("abc"))    # 0

12-5. bool との変換（注意点）

bool → 文字列

flag = True
text = str(flag)

print(text)  # "True"

文字列 → bool（注意）

bool("False")  # True（非空文字列だから）

def to_bool(value: str) -> bool:
    return value.lower() in ("true", "1", "yes")

12-6. FastAPI / Pydantic との関係

from pydantic import BaseModel

class User(BaseModel):
    age: int
    height: float

FastAPI では、リクエストJSONの文字列を
Pydantic が自動で数値に変換してくれます。

13. Python の文字列フォーマット（sprintf / String.format 相当）

13-1. f-string（最推奨・実務標準）

name = "Alice"
age = 20

text = f"name={name}, age={age}"
print(text)

# 出力
name=Alice, age=20

• Python 3.6 以降対応
• 型を意識せずそのまま埋め込める
• 最も読みやすい

13-2. 数値フォーマット（桁数・小数）

小数点以下の桁数指定

value = 3.14159
text = f"{value:.2f}"
print(text)  # 3.14

ゼロ埋め

num = 7
text = f"{num:03d}"
print(text)  # 007

カンマ区切り

price = 1234567
text = f"{price:,}"
print(text)  # 1,234,567

13-3. 文字列幅・寄せ

name = "Bob"

print(f"|{name:<10}|")  # 左寄せ
print(f"|{name:>10}|")  # 右寄せ
print(f"|{name:^10}|")  # 中央寄せ

13-4. 式を直接書ける（f-stringの強み）

x = 10
y = 3

text = f"{x} / {y} = {x / y:.2f}"
print(text)

13-5. format()（旧来方式・参考）

text = "name={}, age={}".format("Alice", 20)
print(text)

Java の String.format に近い感覚ですが、
新規コードでは f-string 推奨です。

13-6. % フォーマット（C sprintf 互換・非推奨）

name = "Alice"
age = 20

text = "name=%s, age=%d" % (name, age)
print(text)

13-7. ログ出力での使い分け（重要）

# OK（f-string）
logger.info(f"user_id={user_id}, status={status}")

# logging 推奨（遅延評価）
logger.info("user_id=%s, status=%s", user_id, status)

13-8. まとめ

• C / Java の sprintf / format 相当は f-string
• 数値・桁・幅指定も全て可能
• 新規コードは f-string 一択でOK

15. datetime（日付・時刻処理）

15-1. 現在時刻を取得する

from datetime import datetime, timezone

now_local = datetime.now()                 # ローカル時刻（タイムゾーンなし）
now_utc   = datetime.now(timezone.utc)     # UTC（タイムゾーンあり）

print(now_local)
print(now_utc)

15-2. 文字列 ⇔ datetime（ISO 8601）

datetime → 文字列

from datetime import datetime, timezone

dt = datetime.now(timezone.utc)
text = dt.isoformat()   # 例: 2026-02-17T12:34:56.123456+00:00
print(text)

文字列 → datetime

from datetime import datetime

text = "2026-02-17T12:34:56+00:00"
dt = datetime.fromisoformat(text)
print(dt)

15-3. 任意フォーマット（strftime / strptime）

datetime → 文字列（strftime）

from datetime import datetime

dt = datetime(2026, 2, 17, 9, 30, 0)
print(dt.strftime("%Y/%m/%d %H:%M:%S"))  # 2026/02/17 09:30:00

文字列 → datetime（strptime）

from datetime import datetime

text = "2026/02/17 09:30:00"
dt = datetime.strptime(text, "%Y/%m/%d %H:%M:%S")
print(dt)

15-4. 日付の加減算（timedelta）

from datetime import datetime, timedelta

dt = datetime(2026, 2, 17, 9, 0, 0)
tomorrow = dt + timedelta(days=1)
after_90m = dt + timedelta(minutes=90)

print(tomorrow)
print(after_90m)

15-5. タイムゾーン（ZoneInfo）

from datetime import datetime
from zoneinfo import ZoneInfo

tokyo = ZoneInfo("Asia/Tokyo")
dt_tokyo = datetime.now(tokyo)

print(dt_tokyo)

16. None / 空文字 / 0 の扱い（truthy / falsy）

16-1. falsy（条件式で False 扱いになる代表）

• None
• ""（空文字）
• 0, 0.0
• [], {}, set()（空コレクション）
• False

values = [None, "", 0, [], {}, set(), False]

for v in values:
    if v:
        print("truthy", v)
    else:
        print("falsy ", v)

16-2. None 判定は is を使う（最重要）

x = None

if x is None:
    print("x is None")

if x is not None:
    print("x has value")

16-3. 「None と空文字と 0」を区別したい例

def classify(x):
    if x is None:
        return "None"
    if x == "":
        return "empty string"
    if x == 0:
        return "zero"
    return "other"

print(classify(None))  # None
print(classify(""))    # empty string
print(classify(0))     # zero

16-4. 実務でよくあるNG例（0が消える）

# NG：0 も False 扱いで消える
age = 0
if not age:
    print("age is missing")  # ← 本当は age=0 が正しいかもしれない

OK（意図を明確にする）：

age = 0
if age is None:
    print("age is missing")

19. 配列（tuple）・list・set・dict の作成方法と使い方

19-1. 配列（固定長っぽい用途）: tuple

tuple は イミュータブル（変更不可） な配列です。
「要素数・内容を変えたくない」用途で使います。

作成方法

# 括弧で作成
point = (10, 20)

# 括弧省略も可
color = 255, 128, 0

# 要素1個の tuple（カンマ必須）
single = (10,)

取り出し

x = point[0]
y = point[1]
print(x, y)

for による取り出し

for v in point:
    print(v)

分解代入（tuple の強み）

x, y = point
print(x, y)

19-2. list（可変長の配列）

作成方法

nums = [1, 2, 3]
names = list(["Alice", "Bob"])

追加・変更・削除

nums.append(4)      # 追加
nums[0] = 10        # 変更
nums.remove(2)      # 値で削除
last = nums.pop()   # 末尾を取り出して削除

for（foreach）で取り出す

for n in nums:
    print(n)

インデックス付き

for i, n in enumerate(nums):
    print(i, n)

19-3. set（集合：重複なし）

作成方法

names = {"Alice", "Bob", "Alice"}   # 重複は自動削除
empty = set()                       # 空集合（{} は dict）

追加・削除

names.add("Carol")
names.discard("Bob")

for で取り出す

for name in names:
    print(name)

集合演算

a = {"A", "B", "C"}
b = {"B", "C", "D"}

print(a | b)  # 和集合
print(a & b)  # 積集合
print(a - b)  # 差集合

19-4. dict（map：キー → 値）

作成方法

user = {
    "id": 1,
    "name": "Alice"
}

user2 = dict(id=2, name="Bob")

追加・更新・削除

user["age"] = 20
user["name"] = "Alicia"
del user["age"]

取り出し（for）

キーだけ：

for k in user:
    print(k)

キーと値：

for k, v in user.items():
    print(k, v)

安全に取得

age = user.get("age")        # 無ければ None
age2 = user.get("age", 0)    # 無ければ 0

19-5. 作成・用途まとめ

17. コレクション（配列 / list / set / map(dict)）の基本・取り出し・相互変換

17-1. Python に存在する代表的なコレクション

17-2. list（順序あり）

作成・追加・削除

nums = [1, 2, 3]
nums.append(4)      # 末尾追加
nums.insert(0, 0)   # 先頭へ挿入
nums.remove(2)      # 値 2 を削除（最初の1つ）
last = nums.pop()   # 末尾を取り出して削除

print(nums)
print(last)

取り出し（for = foreach）

users = ["Alice", "Bob", "Carol"]

for u in users:
    print(u)

インデックス付き（enumerate）

for i, u in enumerate(users):
    print(i, u)

17-3. tuple（固定・変更しない）

point = (10, 20)
x, y = point  # 分解代入
print(x, y)

17-4. set（重複を消す / 集合演算）

names = ["Alice", "Bob", "Alice", "Carol"]
unique = set(names)       # 重複除去
print(unique)

集合演算

a = {"A", "B", "C"}
b = {"B", "C", "D"}

print(a | b)  # 和集合
print(a & b)  # 積集合
print(a - b)  # 差集合

取り出し（for）

for x in unique:
    print(x)

17-5. dict（map：キー→値）

作成・追加・更新・削除

user = {"id": 1, "name": "Alice"}

user["age"] = 20           # 追加
user["name"] = "Alicia"    # 更新
del user["age"]            # 削除

print(user)

取り出し（キーだけ）

for k in user:
    print(k)  # key

キーと値（items）

for k, v in user.items():
    print(k, v)

安全に取得（get）

age = user.get("age")          # 無ければ None
age2 = user.get("age", 0)      # 無ければ 0
print(age, age2)

17-6. 相互変換（list / set / tuple / dict）

list → set（重複除去）

nums = [1, 2, 2, 3]
unique = set(nums)
print(unique)  # {1, 2, 3}

set → list（並べ替えてから使うのが定石）

unique = {3, 1, 2}
sorted_list = sorted(unique)
print(sorted_list)  # [1, 2, 3]

list ↔ tuple

nums = [1, 2, 3]
t = tuple(nums)
l = list(t)
print(t, l)

dict → list（キー、値、キー値ペア）

user = {"id": 1, "name": "Alice"}

keys = list(user.keys())
values = list(user.values())
pairs = list(user.items())

print(keys)
print(values)
print(pairs)

list（ペア）→ dict

pairs = [("id", 1), ("name", "Alice")]
user = dict(pairs)
print(user)

17-7. よく使う標準メソッドまとめ

list

• append(x)：末尾追加
• extend(iterable)：まとめて追加
• insert(i, x)：挿入
• remove(x)：値で削除
• pop()：末尾取り出し＋削除
• sort()：破壊的ソート（戻り値なし）

set

• add(x)：追加
• remove(x)：削除（無いと例外）
• discard(x)：削除（無くてもOK）

dict

• get(key, default)：安全取得
• keys(), values(), items()
• pop(key)：取り出し＋削除
• update(other)：マージ

7. ミュータブル / イミュータブル完全理解

7-1. ミュータブル / イミュータブルとは

7-2. イミュータブルの例

a = 10
b = a
b += 1

print(a)  # 10
print(b)  # 11

b += 1 は 新しい int オブジェクト を作るだけ。
a は一切影響を受けません。

7-3. ミュータブルの例（危険）

a = [1, 2]
b = a
b.append(3)

print(a)  # [1, 2, 3]
print(b)  # [1, 2, 3]

7-4. 実務での鉄則

• 共有される可能性がある場所では ミュータブルを直接持たない
• 必要なら copy() / deepcopy() を使う
• 状態管理は DB / クラス / Pydantic に寄せる

4. Python の演算子まとめ

4-1. 算術演算子

4-2. 論理演算子

if a > 0 and b < 10:
    print("OK")

if not is_valid:
    print("NG")

4-3. ビット演算子

a = 0b1010
b = 0b1100

print(a & b)  # AND
print(a | b)  # OR
print(a ^ b)  # XOR

4-4. 代入演算子（網羅）

4-2. 比較演算子（条件判定の基本）

4-5. 三項演算子（条件式）

result = "OK" if status == 200 else "NG"

5. Python で正規表現（Regex）を扱う方法（実務向け）

5-1. 基本的な使い方

import re

pattern = r"\d{4}-\d{2}-\d{2}"
text = "2026-02-01"

if re.match(pattern, text):
    print("日付形式OK")

5-2. よく使う関数

5-3. 実務で多い例

メールアドレスチェック（簡易）

email_pattern = r"^[\w\.-]+@[\w\.-]+\.\w+$"

if re.match(email_pattern, email):
    print("valid")

ログからID抽出

log = "user_id=12345 action=login"
m = re.search(r"user_id=(\d+)", log)

if m:
    user_id = m.group(1)

5-4. re.compile を使う理由（重要）

pattern = re.compile(r"\d+")

pattern.findall("id=10 id=20")

10. Python の命名規則（クラス・関数・変数・定数・モジュール）

10-1. 命名規則の全体像

10-2. クラス命名（PascalCase）

class UserService:
    pass

class OrderStatus:
    pass

• 名詞・役割名で表す
• 動詞は使わない
• 1クラス1責務が基本

10-3. 関数・メソッド命名（snake_case）

def get_user(user_id: int):
    pass

def create_order(data):
    pass

• 動詞 + 目的語
• 副作用がある場合も同じルール
• FastAPI のエンドポイント名とも一致させると分かりやすい

10-4. 変数命名（snake_case）

user_name = "Alice"
is_active = True
order_count = 3

• 意味が分かる名前を付ける
• tmp, data 連発は避ける
• 真偽値は is_, has_ で始める

10-5. 定数命名（UPPER_SNAKE_CASE）

API_TIMEOUT = 30
DEFAULT_PAGE_SIZE = 20

10-6. インターフェース相当（Protocol / ABC）

Protocol（推奨・Pythonic）

from typing import Protocol

class UserRepository(Protocol):
    def find_by_id(self, user_id: int): ...

ABC（抽象基底クラス）

from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int):
        pass

• 命名はクラスと同じ PascalCase
• ~Interface という接尾辞は 付けない のが一般的

10-7. 非公開（internal）を表す命名

def _internal_function():
    pass

class Service:
    def _helper(self):
        pass

• _ 1個：内部利用（慣習）
• __ 2個：名前マングリング（特殊用途）

10-8. FastAPI 実務向け命名指針

10-9. Java / PHP との違いまとめ

11. Python のファイル分け（モジュール分割の考え方）

11-1. Python におけるファイル分けの基本思想

• .py ファイル ＝ モジュール
• ファイル名は snake_case
• 「何の責務か」が分かる単位で分ける

11-2. インターフェース相当のファイル分け

Python には interface キーワードはありません。
実務では以下のどちらかを使い、専用ファイルに分けます。

Protocol（推奨）

# app/repositories/user_repository.py
from typing import Protocol

class UserRepository(Protocol):
    def find_by_id(self, user_id: int): ...

• 役割（契約）だけを書く
• 実装は別ファイルに分離
• 型チェック（mypy / Pylance）で効く

11-3. 抽象クラス（ABC）のファイル分け

# app/services/base_service.py
from abc import ABC, abstractmethod

class BaseService(ABC):

    @abstractmethod
    def execute(self):
        pass

• 共通処理＋未実装メソッドを定義
• 継承される前提のクラス
• 1ファイルに1抽象クラスが分かりやすい

11-4. 具象（public）クラスのファイル分け

# app/services/user_service.py
from app.repositories.user_repository import UserRepository

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def get_user(self, user_id: int):
        return self.repo.find_by_id(user_id)

• アプリの実処理を担うクラス
• 外部から使われる「publicクラス」
• 原則 1クラス1ファイル

11-5. 実装クラス（Repository / Adapter）の分け方

# app/repositories/user_repository_impl.py
from app.repositories.user_repository import UserRepository

class UserRepositoryImpl(UserRepository):
    def find_by_id(self, user_id: int):
        return {"id": user_id, "name": "Alice"}

11-6. 関数中心ファイルの分け方

# app/utils/date_utils.py
def parse_date(value: str):
    ...

def format_date(dt):
    ...

• クラスにするほどでもない処理
• 純粋関数が中心
• 副作用を持たせない

11-7. ファイル分けの実務テンプレ（FastAPI）

app/
  routers/
    users.py          # API入口
  services/
    user_service.py   # 業務ロジック
  repositories/
    user_repository.py        # Protocol
    inmemory_user_repository.py
  schemas/
    user.py           # Pydantic
  core/
    config.py

11-8. やりがちなアンチパターン

11-9. 判断基準（迷ったら）

20. 関数・メソッドの定義と使い方（戻り値・引数・ラムダ）

20-1. 関数の定義方法

def add(a, b):
    return a + b

result = add(2, 3)
print(result)  # 5

• def キーワードで定義
• インデントがブロック
• 型指定は必須ではない

20-2. メソッドの定義方法（クラス内）

class Calculator:
    def add(self, a, b):
        return a + b

calc = Calculator()
print(calc.add(2, 3))

20-3. 戻り値（return）の指定

1つの値を返す

def get_name():
    return "Alice"

複数の値を返す（tuple）

def get_point():
    return 10, 20   # tuple を返す

x, y = get_point()
print(x, y)

return が無い場合

def do_nothing():
    pass

result = do_nothing()
print(result)  # None

20-4. 戻り値の型指定（型ヒント）

def add(a: int, b: int) -> int:
    return a + b

• 実行時に強制はされない
• IDE補完・静的解析（mypy）で効果あり

20-5. 引数の種類

① 通常の引数

def greet(name):
    print(f"Hello {name}")

② 初期値（デフォルト引数）

def greet(name="Guest"):
    print(f"Hello {name}")

③ 可変長引数（*args）

def total(*args):
    return sum(args)

print(total(1, 2, 3))  # 6

*args は tuple として受け取ります。

④ キーワード引数（**kwargs）

def print_user(**kwargs):
    for k, v in kwargs.items():
        print(k, v)

print_user(name="Alice", age=20)

**kwargs は dict として受け取ります。

⑤ 名前付き引数（キーワード引数）

def connect(host, port):
    print(host, port)

connect(port=5432, host="localhost")

20-6. 引数の順序ルール（重要）

def func(a, b=0, *args, c=1, **kwargs):
    pass

引数の順序は以下の通りです：

1. 通常引数
2. デフォルト引数
3. *args
4. キーワード専用引数
5. **kwargs

20-7. ラムダ関数（無名関数）

# 通常の関数
def square(x):
    return x * x

# ラムダ関数
square_lambda = lambda x: x * x

print(square(3))
print(square_lambda(3))

よく使う場面（sorted / map / filter）

nums = [1, 2, 3, 4]

# key にラムダ
sorted_nums = sorted(nums, key=lambda x: -x)

# map
squares = list(map(lambda x: x * x, nums))

# filter
evens = list(filter(lambda x: x % 2 == 0, nums))

20-8. まとめ

• def で関数・メソッドを定義
• 戻り値は return（無ければ None）
• 引数は柔軟だが順序に注意
• ラムダは補助的に使う

21. with 文（リソース管理）

# with を使わない場合（危険）
f = open("data.txt", "r")
content = f.read()
f.close()

# with を使う場合（安全）
with open("data.txt", "r") as f:
    content = f.read()
# ここで自動的に close() される

with open("data.txt", "r") as f:
    for line in f:
        print(line)
        raise RuntimeError("エラー発生")  # それでも close される

# DB接続のイメージ
with db.connect() as conn:
    conn.execute("SELECT * FROM users")

class Resource:
    def __enter__(self):
        print("open")
        return self

    def __exit__(self, exc_type, exc, tb):
        print("close")

with Resource():
    print("using resource")

21. Javaの interface / abstract class / class / enum / record 相当を Python でどう定義するか（違いとサンプル）

21-1. interface 相当：Protocol（推奨） / ABC（厳格）

① Protocol（Pythonic・実務推奨：構造的サブタイピング）

from typing import Protocol

class UserRepository(Protocol):
    """
    Javaのinterface相当。
    実装クラスが '同じメソッドを持っていれば' それでOK（duck typing + 型チェック）。
    """
    def find_by_id(self, user_id: int) -> dict: ...

class InMemoryUserRepository:
    """interfaceをimplementsしなくても、メソッドがあればOK。"""
    def find_by_id(self, user_id: int) -> dict:
        return {"id": user_id, "name": "Alice"}

② ABC（抽象基底クラス：実行時にもある程度“強制”したい場合）

from abc import ABC, abstractmethod

class UserRepositoryABC(ABC):
    """
    Javaのinterface + abstract method に近い。
    抽象メソッドを未実装だとインスタンス化できない。
    """
    @abstractmethod
    def find_by_id(self, user_id: int) -> dict:
        pass

21-2. abstract class：ABC を使う

from abc import ABC, abstractmethod

class BaseService(ABC):
    """
    Javaのabstract class相当。
    共通処理（具体メソッド）と、子で実装すべき抽象メソッドを混在できる。
    """
    def log_start(self) -> None:
        print("start")  # 共通処理（具体メソッド）

    @abstractmethod
    def execute(self) -> dict:
        """子クラスで必ず実装してほしい処理"""
        pass

class UserService(BaseService):
    def execute(self) -> dict:
        self.log_start()
        return {"result": "ok"}

21-3. class：通常クラス（最頻出）

class User:
    """
    Javaのclass相当。
    Pythonはフィールド宣言が必須ではなく、__init__ で属性を作る。
    """
    def __init__(self, user_id: int, name: str):
        self.user_id = user_id
        self.name = name

    def greet(self) -> str:
        return f"Hello, {self.name}"

u = User(1, "Alice")
print(u.greet())

21-4. enum：Enum クラス（標準ライブラリ）

from enum import Enum

class OrderStatus(Enum):
    """
    Javaのenum相当。
    値はシンボルとして扱われ、比較は '==' でOK。
    """
    NEW = "new"
    PAID = "paid"
    SHIPPED = "shipped"

status = OrderStatus.NEW

if status == OrderStatus.NEW:
    print("new order")

print(status.value)  # "new"

21-5. record 相当：dataclass / NamedTuple（目的別に使い分け）

① dataclass（実務で最も使う“データの入れ物”）

from dataclasses import dataclass

@dataclass(frozen=True)
class UserRecord:
    """
    Java record に近い用途。
    frozen=True にすると不変（イミュータブル）にできる。
    """
    user_id: int
    name: str

u = UserRecord(1, "Alice")
print(u.user_id, u.name)
# u.name = "Bob"  # frozen=True ならエラー（不変）

② NamedTuple（より“タプル寄り”の不変データ）

from typing import NamedTuple

class Point(NamedTuple):
    """
    recordというより「名前付きtuple」。
    軽量で不変、分解代入も得意。
    """
    x: int
    y: int

p = Point(10, 20)
print(p.x, p.y)

x, y = p  # 分解代入
print(x, y)

21-6. FastAPI視点：Pydantic（record“っぽい”が目的が違う）

from pydantic import BaseModel

class UserResponse(BaseModel):
    """
    Java record のように「データの入れ物」に見えるが、目的は違う。
    API入力/出力の型保証とバリデーションが主目的。
    """
    id: int
    name: str
    age: int | None = None

# dict -> Pydantic -> JSON（FastAPIが自動）
user = UserResponse(id=1, name="Alice", age=20)
print(user.model_dump())  # dict化（Pydantic v2）

21-7. まとめ

• Pythonは「構文で縛る」より「慣習と型ヒントで整える」文化
• interface相当は Protocol が最も実務的
• record相当は dataclass（不変なら frozen=True）
• FastAPIでは Pydantic が “API用データ型” の主役

22. Protocol（構造型）：継承しなくても「形が合えばOK」とはどういう意味か

22-1. 前提：Protocol（契約）を定義する

from typing import Protocol

class UserRepository(Protocol):
    """
    Javaのinterface相当。
    「find_by_id(int) -> dict を持っていること」が契約。
    """
    def find_by_id(self, user_id: int) -> dict:
        ...

この時点では、
「UserRepository を継承しなければならない」 とは 一切書かれていません。

22-2. 継承していないが「形が合っている」例（構造型）

class InMemoryUserRepository:
    """
    UserRepository を継承していない。
    しかし、同じ名前・引数・戻り値のメソッドを持っている。
    """
    def find_by_id(self, user_id: int) -> dict:
        return {"id": user_id, "name": "Alice"}

def get_user_name(repo: UserRepository, user_id: int) -> str:
    """
    引数 repo は UserRepository として扱われる。
    """
    user = repo.find_by_id(user_id)
    return user["name"]


repo = InMemoryUserRepository()
print(get_user_name(repo, 1))  # Alice

22-3. 継承している例（名義的に契約を示す）

class SqlUserRepository(UserRepository):
    """
    UserRepository を明示的に継承している。
    Javaの implements に近い書き方。
    """
    def find_by_id(self, user_id: int) -> dict:
        return {"id": user_id, "name": "Bob"}

repo = SqlUserRepository()
print(get_user_name(repo, 1))  # Bob

22-4. 「形が合っていない」とどうなるか

class BadRepository:
    """
    メソッド名が契約と違う。
    """
    def find(self, user_id: int) -> dict:
        return {"id": user_id}

# repo = BadRepository()
# get_user_name(repo, 1)
# 実行時：AttributeError（find_by_id が無い）

• 実行時：find_by_id が無いためエラー
• 型チェック（IDE / mypy）：事前に警告できる

22-5. Javaとの決定的な違い

22-6. なぜ Python はこの設計なのか

• Pythonは元々 duck typing の言語
• Protocolはそれを「型ヒントとして明文化」したもの
• テスト用の Fake / Mock を簡単に差し替えられる
• FastAPI の DI（Depends）と非常に相性が良い

23. Python における「構造体（struct）」相当の考え方

23-1. dataclass（最も一般的な構造体相当）

dataclass は「データを保持するだけのクラス」を
簡単・安全に定義するための仕組みです。
実務では struct の代わりにこれを使うのが基本です。

from dataclasses import dataclass

@dataclass
class User:
    """
    Cのstruct / JavaのDTO・record相当
    """
    id: int
    name: str
    age: int

u = User(1, "Alice", 20)
print(u.id, u.name, u.age)

• フィールド定義が一目で分かる
• __init__, __repr__ 等が自動生成
• ミュータブル（書き換え可能）

不変（immutable）な構造体にする

@dataclass(frozen=True)
class Point:
    x: int
    y: int

p = Point(10, 20)
# p.x = 30  # エラー（不変）

23-2. NamedTuple（軽量・不変な構造体）

NamedTuple は「名前付き tuple」です。
軽量・高速・不変という特徴があります。

from typing import NamedTuple

class Point(NamedTuple):
    x: int
    y: int

p = Point(10, 20)
print(p.x, p.y)

x, y = p   # 分解代入
print(x, y)

• イミュータブル（変更不可）
• tuple と同じ振る舞い
• 構造体というより「値オブジェクト」

23-3. SimpleNamespace（動的な構造体）

実行時にフィールドを自由に追加したい場合は
SimpleNamespace が使えます。

from types import SimpleNamespace

user = SimpleNamespace()
user.id = 1
user.name = "Alice"
user.age = 20

print(user.name)

• 超動的（フィールド定義なし）
• 型安全性は低い
• 一時的なデータまとめ用途向け

23-4. dict（最も原始的な構造体）

user = {
    "id": 1,
    "name": "Alice",
    "age": 20
}

• 柔軟だが型保証なし
• キーの打ち間違いに弱い
• 構造体というより「生データ」

23-5. ctypes.Structure（C互換struct・特殊用途）

C言語の struct と メモリレイアウト互換が必要な場合、
ctypes.Structure を使います。

from ctypes import Structure, c_int, c_char_p

class CUser(Structure):
    _fields_ = [
        ("id", c_int),
        ("name", c_char_p),
    ]

• Cライブラリ連携専用
• 通常の業務アプリではほぼ使わない

23-6. どれを使うべきか（まとめ）

23. Python における「構造体（struct）」相当の考え方

23-1. dataclass（最も一般的な構造体相当）

dataclass は「データを保持するだけのクラス」を
簡単・安全に定義するための仕組みです。
実務では struct の代わりにこれを使うのが基本です。

from dataclasses import dataclass

@dataclass
class User:
    """
    Cのstruct / JavaのDTO・record相当
    """
    id: int
    name: str
    age: int

u = User(1, "Alice", 20)
print(u.id, u.name, u.age)

• フィールド定義が一目で分かる
• __init__, __repr__ 等が自動生成
• ミュータブル（書き換え可能）

不変（immutable）な構造体にする

@dataclass(frozen=True)
class Point:
    x: int
    y: int

p = Point(10, 20)
# p.x = 30  # エラー（不変）

23-2. NamedTuple（軽量・不変な構造体）

NamedTuple は「名前付き tuple」です。
軽量・高速・不変という特徴があります。

from typing import NamedTuple

class Point(NamedTuple):
    x: int
    y: int

p = Point(10, 20)
print(p.x, p.y)

x, y = p   # 分解代入
print(x, y)

• イミュータブル（変更不可）
• tuple と同じ振る舞い
• 構造体というより「値オブジェクト」

23-3. SimpleNamespace（動的な構造体）

実行時にフィールドを自由に追加したい場合は
SimpleNamespace が使えます。

from types import SimpleNamespace

user = SimpleNamespace()
user.id = 1
user.name = "Alice"
user.age = 20

print(user.name)

• 超動的（フィールド定義なし）
• 型安全性は低い
• 一時的なデータまとめ用途向け

23-4. dict（最も原始的な構造体）

user = {
    "id": 1,
    "name": "Alice",
    "age": 20
}

• 柔軟だが型保証なし
• キーの打ち間違いに弱い
• 構造体というより「生データ」

23-5. ctypes.Structure（C互換struct・特殊用途）

C言語の struct と メモリレイアウト互換が必要な場合、
ctypes.Structure を使います。

from ctypes import Structure, c_int, c_char_p

class CUser(Structure):
    _fields_ = [
        ("id", c_int),
        ("name", c_char_p),
    ]

• Cライブラリ連携専用
• 通常の業務アプリではほぼ使わない

23-6. どれを使うべきか（まとめ）

Pythonにおけるイテレータとジェネレータ

1. まず結論（超重要）

2. イテレータとは何か

イテレータとは、
「次の要素を1つずつ返すことができるオブジェクト」です。

2-1. イテレータの条件

• __iter__() メソッドを持つ
• __next__() メソッドを持つ
• 要素が無くなったら StopIteration を送出する

2-2. listはイテレータ？

nums = [1, 2, 3]

it = iter(nums)      # イテレータを取得
print(next(it))      # 1
print(next(it))      # 2
print(next(it))      # 3
# print(next(it))    # StopIteration

3. 自作イテレータ（クラスで書く場合）

Javaでいう Iterator を自分で実装する感覚に近いです。

# 0 から max-1 まで返すイテレータ
class CounterIterator:
    def __init__(self, max_value: int):
        self.max = max_value
        self.current = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.current >= self.max:
            raise StopIteration
        value = self.current
        self.current += 1
        return value

for i in CounterIterator(3):
    print(i)
# 0, 1, 2

4. ジェネレータとは何か

ジェネレータは、
yield を使って「途中で止まりながら値を返す関数」です。

def counter(max_value: int):
    current = 0
    while current < max_value:
        yield current
        current += 1

for i in counter(3):
    print(i)
# 0, 1, 2

5. ジェネレータの動作イメージ

1. 関数呼び出し → ジェネレータオブジェクトが返る
2. next() が呼ばれる
3. yield まで処理が進む
4. 値を返して 状態を保持したまま停止
5. 次の next() で続きから再開

6. ジェネレータ式（generator expression）

list内包表記とほぼ同じ見た目ですが、
() を使うとジェネレータになります。

# list（すべてメモリに載る）
lst = [i * 2 for i in range(5)]

# generator（1件ずつ生成）
gen = (i * 2 for i in range(5))

print(lst)        # [0, 2, 4, 6, 8]
print(next(gen))  # 0
print(next(gen))  # 2

7. メモリ効率の違い（実務で重要）

# 巨大ファイルの行を処理する例

# 悪い例（全行をメモリに読む）
lines = open("big.txt").readlines()
for line in lines:
    process(line)

# 良い例（1行ずつ処理）
def read_lines(path):
    with open(path) as f:
        for line in f:
            yield line

for line in read_lines("big.txt"):
    process(line)

8. イテレータ vs ジェネレータ まとめ

9. Java経験者向け一言まとめ

sorted / key / lambda の使い方（Python並び替えの完全理解）

1. まず結論（ここだけ読めば8割分かる）

2. sorted() の基本

sorted() は 元のデータを壊さず、
並び替えた 新しい list を返します。

nums = [5, 1, 3, 2]
result = sorted(nums)

print(result)  # [1, 2, 3, 5]
print(nums)    # [5, 1, 3, 2]（元のまま）

3. key とは何か（ここが一番大事）

key は 「各要素から、並び替え用の値を取り出す関数」です。

3-1. keyなしの場合

words = ["banana", "apple", "cherry"]
print(sorted(words))
# ['apple', 'banana', 'cherry']

→ 要素そのもの（文字列）が比較される

3-2. keyを使う場合（文字列長でソート）

words = ["banana", "apple", "cherry"]

result = sorted(words, key=len)
print(result)
# ['apple', 'banana', 'cherry']

4. lambda を使った key の書き方

lambda は 1行で書ける無名関数です。

lambda x: x * 2
# ↑ 引数 x を受け取って x*2 を返す関数

4-1. dict の特定キーで並び替える

users = [
    {"id": 3, "name": "Bob"},
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Carol"},
]

result = sorted(users, key=lambda u: u["id"])
print(result)

→ id の昇順で並ぶ

4-2. オブジェクト（Entity / DTO）で並び替える

# UserEntity(id, name)
result = sorted(entities, key=lambda e: e.id)

→ e.id を基準に並ぶ

5. 複数条件で並び替える（超実務）

Pythonでは タプル を返せば、
左から順に比較されます。

# age 昇順 → name 昇順
result = sorted(users, key=lambda u: (u["age"], u["name"]))

6. 降順で並び替える

# id を降順
result = sorted(users, key=lambda u: u["id"], reverse=True)

7. key に関数名を渡す vs lambda

def get_id(u):
    return u["id"]

sorted(users, key=get_id)      # OK
sorted(users, key=lambda u: u["id"])  # もっと一般的

8. よくある勘違い（初心者ポイント）

• key は「比較関数」ではない（Javaと違う）
• lambda は1回も比較を担当しない
• Pythonは「一度 key を計算 → その結果で並び替え」

9. Java経験者向け対応表

10. 実務でのまとめ

24. Pydantic の概要と使い方 → dataclass との変換（相互変換）

24-1. Pydantic とは（概要・目的）

Pydantic は「型ヒント」をもとに、
データのバリデーション（検証）と変換を行うライブラリです。
FastAPI の Request / Response は Pydantic で表現するのが定石です。

• 目的1：API入力を安全に受け取る（型不正・欠損を弾く）
• 目的2：API出力を安全に返す（余計な情報漏えいを防ぐ）
• 目的3：dict/JSON と相互変換する（model_dump() 等）

24-2. Pydantic の基本的な使い方（定義・生成・バリデーション）

from pydantic import BaseModel

class UserRequest(BaseModel):
    """
    APIの入力（例：POST /users のリクエストBody）
    """
    name: str
    age: int | None = None

生成（dict から）

payload = {"name": "Alice", "age": "20"}  # ageが文字列でも…
user_req = UserRequest(**payload)

print(user_req.name)  # Alice
print(user_req.age)   # 20 （intに変換される）

24-3. FastAPI での利用例（呼び出し例）

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserResponse(BaseModel):
    id: int
    name: str
    age: int | None = None

@app.post("/users", response_model=UserResponse)
def create_user(req: UserRequest):
    # req は既にバリデーション済み
    # 本来はDB登録してIDを発行するが、ここではダミー
    return UserResponse(id=1, name=req.name, age=req.age)

FastAPI は リクエストJSON → Pydantic → Pythonオブジェクト、
Pydantic → JSONレスポンス を自動で行います。

24-4. Pydantic と dataclass の役割分担（実務）

24-5. dataclass の定義（内部用データ）

from dataclasses import dataclass

@dataclass
class UserEntity:
    """
    内部処理用（DB/サービス層）
    """
    id: int
    name: str
    age: int | None = None

24-6. 変換①：Pydantic → dataclass（API入力を内部データへ）

API入力（Pydantic）を、内部データ（dataclass）へ渡すときの定番は
Pydanticをdict化 → dataclassへ詰め替えです。

def to_entity(req: UserRequest, new_id: int) -> UserEntity:
    data = req.model_dump()        # Pydantic v2: dict化
    return UserEntity(id=new_id, **data)

req = UserRequest(name="Alice", age=20)
entity = to_entity(req, new_id=100)

print(entity)  # UserEntity(id=100, name='Alice', age=20)

24-7. 変換②：dataclass → Pydantic（内部データをAPI出力へ）

DBや業務ロジックで作った dataclass を、レスポンス用 Pydantic に変換します。

from dataclasses import asdict

def to_response(entity: UserEntity) -> UserResponse:
    data = asdict(entity)            # dataclass → dict
    return UserResponse(**data)      # dict → Pydantic

entity = UserEntity(id=100, name="Alice", age=20)
res = to_response(entity)

print(res.model_dump())  # {'id': 100, 'name': 'Alice', 'age': 20}

24-8. 変換③：リスト（複数件）の変換例

entities = [
    UserEntity(id=1, name="Alice", age=20),
    UserEntity(id=2, name="Bob", age=None),
]

responses = [UserResponse(**asdict(e)) for e in entities]
print([r.model_dump() for r in responses])

24-9. 実務での落とし穴と対策

24-10. まとめ

• Pydantic は API境界での 型保証・バリデーション・JSON変換の主役
• dataclass は 内部処理の 軽量なデータの入れ物として便利
• 変換は基本：Pydantic(model_dump) → dict → dataclass
• 逆変換は：dataclass(asdict) → dict → Pydantic

25. Python（FastAPI）における DTO / Entity / Collection / Service / Repository の概念と Spring Boot との違い

25-1. 各概念の役割（対応表）

25-2. DTO（FastAPIではPydanticが主役）

Spring Boot では DTO を自作クラスで作り、Jackson が JSON 変換します。
FastAPI では DTO に相当するものは Pydanticモデルです（型保証＋検証＋JSON変換が一体）。

from pydantic import BaseModel

class UserCreateRequest(BaseModel):
    """
    DTO（Request）相当:
    クライアントから来るJSONを受け取る型
    """
    name: str
    age: int | None = None

class UserResponse(BaseModel):
    """
    DTO（Response）相当:
    クライアントへ返すJSONの型
    """
    id: int
    name: str
    age: int | None = None

25-3. Entity（DBモデル / ドメインモデル）

Spring Boot では JPA Entity が中心になりがちです。
Python では用途で分かれます：

• DBと直結するモデル（ORM）：SQLAlchemyモデルなど
• 業務ロジック用の内部モデル：dataclass（Entity的に扱う）

from dataclasses import dataclass

@dataclass
class UserEntity:
    """
    Entity相当（内部モデル）:
    DB由来でもよいし、業務ロジック用に整形した形でもよい。
    """
    id: int
    name: str
    age: int | None = None

25-4. Repository（データアクセス層：Protocolで契約を作る）

Spring Boot は JpaRepository が強力で、DIで自動注入されます。
Python では Protocol（構造型）で「契約」を作り、実装を差し替えられるようにするのが実務的です。

from typing import Protocol

class UserRepository(Protocol):
    """
    Repository契約（interface相当）:
    find/save など「データアクセスAPI」を定義する。
    """
    def find_by_id(self, user_id: int) -> UserEntity | None: ...
    def save(self, user: UserEntity) -> UserEntity: ...

class InMemoryUserRepository:
    """
    Repository実装例（テストや試作で便利）:
    Protocolを継承しなくても「形が合えばOK」。
    """
    def __init__(self):
        self._store: dict[int, UserEntity] = {}

    def find_by_id(self, user_id: int) -> UserEntity | None:
        return self._store.get(user_id)

    def save(self, user: UserEntity) -> UserEntity:
        self._store[user.id] = user
        return user

25-5. Service（業務ロジック層）

Spring の @Service と同じく、Pythonでも Service は「業務ロジックの中心」です。
ただし Python はアノテーションではなく 普通のクラスとして作ります。

class UserService:
    """
    Service層:
    Controller（Router）から呼ばれ、Repositoryを使って処理する。
    """
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def create_user(self, new_id: int, name: str, age: int | None) -> UserEntity:
        # 例：業務チェック
        if not name:
            raise ValueError("name is required")

        user = UserEntity(id=new_id, name=name, age=age)
        return self.repo.save(user)

25-6. Controller相当（FastAPIではRouter/Endpoint関数）

from fastapi import FastAPI

app = FastAPI()

# DIの代わりに “組み立て” する例（手動注入）
repo = InMemoryUserRepository()
service = UserService(repo)

@app.post("/users", response_model=UserResponse)
def create_user(req: UserCreateRequest):
    """
    Controller相当:
    Springの@RestControllerメソッドに近い役割。
    """
    entity = service.create_user(new_id=1, name=req.name, age=req.age)
    return UserResponse(id=entity.id, name=entity.name, age=entity.age)

25-7. Collection（複数件の扱い：list/set/dict + 内包表記）

Spring Boot では List/Set/Map に Stream を組み合わせることが多いです。
Python では list/set/dict に加えて 内包表記が強力です。

entities = [
    UserEntity(id=1, name="Alice", age=20),
    UserEntity(id=2, name="Bob", age=None),
]

# Collection変換（Entity -> Response DTO）: 内包表記
responses = [UserResponse(id=e.id, name=e.name, age=e.age) for e in entities]

25-8. まとめ：Python（FastAPI）設計の特徴

• 概念（DTO/Entity/Service/Repository）はあるが、強制は少ない
• DTOは Pydantic、Entityは dataclass/ORM、Repositoryは Protocol が相性良い
• DIは “魔法” より “明示” に寄る（追いやすい・テストしやすい）
• 結果として、規模が大きくなるほど「ルール（慣習）を決めて守る」が重要

26. Pydanticモデル

26-1. 概要

Pydanticモデルは、Pythonの型ヒントを使って
データの検証（バリデーション）・型変換・JSON変換を行うためのモデルです。
FastAPIでは DTO（Request / Response） として使われるのが基本です。

26-2. 目的

• API入力の型チェック・必須チェック
• API出力の安全化（返してよい項目だけ定義）
• dict / JSON との相互変換
• OpenAPI（Swagger）仕様を自動生成

26-3. サンプルコード

from pydantic import BaseModel

class UserResponse(BaseModel):
    """
    APIレスポンス用DTO
    """
    id: int
    name: str
    age: int | None = None

# dict から生成
data = {"id": 1, "name": "Alice", "age": "20"}
user = UserResponse(**data)

print(user.age)          # 20（intに変換）
print(user.model_dump()) # dict化（Pydantic v2）

26-4. 詳細説明

• 型が合わない場合は ValidationError
• 未定義のフィールドは自動で弾かれる
• FastAPIでは return UserResponse(...) だけでJSON化
• 「API境界専用モデル」と割り切るのが設計のコツ

27. SQLAlchemyモデル

27-1. 概要

SQLAlchemyモデルは、
Pythonクラスとデータベースのテーブルを対応付ける ORM（Object Relational Mapping） です。

27-2. 目的

• SQLを書かずにDB操作を行う
• DBレコードをPythonオブジェクトとして扱う
• DB差異（SQLite / PostgreSQL等）を吸収

27-3. サンプルコード

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    """
    users テーブルに対応する ORMモデル
    """
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String)
    age: Mapped[int | None] = mapped_column(Integer, nullable=True)

# DBから取得したレコードのイメージ
user = UserModel(id=1, name="Alice", age=20)
print(user.name)

27-4. 詳細説明

• SQLAlchemyモデルは「DB構造寄り」
• そのままAPIレスポンスに使うのは非推奨
• 実務では dataclass や Pydantic に変換して使う
• Spring Boot の JPA Entity に最も近い存在

28. 内包表記（list / dict / set comprehension）

28-1. 概要

内包表記は、
コレクション（list / dict / set）を 簡潔に生成・変換するためのPython構文です。

28-2. 目的

• ループ処理を短く・読みやすく書く
• Collection変換（map / filter 相当）
• DTO変換やレスポンス生成で多用される

28-3. サンプルコード

list内包表記

nums = [1, 2, 3, 4]

squares = [n * n for n in nums]
evens = [n for n in nums if n % 2 == 0]

dict内包表記

users = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
]

user_map = {u["id"]: u["name"] for u in users}

set内包表記

names = ["Alice", "Bob", "Alice"]

unique_names = {name for name in names}

28-4. 詳細説明

• Javaの Stream API の map / filter に相当
• 1行で意味が分かる範囲に留める
• 複雑になったら通常の for に戻すのがPython流
• FastAPIでは「Entity → DTO変換」で頻出

29. FastAPI の DI（依存性注入 / Depends）

29-1. 概要

FastAPI の DI は Depends を使って、
エンドポイント関数（Controller相当）へ 必要なオブジェクトを自動で注入する仕組みです。
Spring Boot の @Autowired のように「コンテナが勝手に注入」するのではなく、
“この引数はDependsで解決する” と明示したものだけがDIされるのが大きな特徴です。

29-2. 目的

• 責務分離：API入口（Router）と業務ロジック（Service）を分ける
• 差し替え容易：本番Repo ↔ テストRepo を差し替えやすい
• 共通処理：認証、DBセッション、ログなどを共通化
• テスト容易：依存を差し替えて単体テストしやすい

29-3. どの場合「DIされる」か（Depends がある場合）

FastAPI で DI されるのは、基本的に以下のケースです：

• 引数が Depends(...) で定義されている
• ルーターやアプリに依存が登録されている（dependencies=[Depends(...)]）

例：依存（DBセッション）をDIする

from fastapi import FastAPI, Depends
from typing import Generator

app = FastAPI()

class DbSession:
    def __init__(self):
        self.connected = True

    def close(self):
        self.connected = False

def get_db() -> Generator[DbSession, None, None]:
    """
    DIで注入される依存（DBセッション）。
    yield を使うと「最後に必ず後始末」できる。
    """
    db = DbSession()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db: DbSession = Depends(get_db)):
    """
    db は FastAPI が get_db() を呼んで注入する。
    """
    return {"user_id": user_id, "db_connected": db.connected}

例：Service をDIする（RepositoryもDIする）

from typing import Protocol

class UserRepository(Protocol):
    def find_name(self, user_id: int) -> str: ...

class InMemoryUserRepository:
    def find_name(self, user_id: int) -> str:
        return "Alice"

def get_repo() -> UserRepository:
    return InMemoryUserRepository()

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def get_name(self, user_id: int) -> str:
        return self.repo.find_name(user_id)

def get_service(repo: UserRepository = Depends(get_repo)) -> UserService:
    """
    Service生成にも Depends を使える（依存の連鎖）。
    """
    return UserService(repo)

@app.get("/names/{user_id}")
def get_name(user_id: int, service: UserService = Depends(get_service)):
    return {"name": service.get_name(user_id)}

29-4. どの場合「DIされない」か（Dependsが無い場合）

Spring Boot ではフィールドに @Autowired が付くなどで注入されますが、
FastAPI は Depends を書かない限り注入しません。

例：Dependsが無いのでDIされない（よくある誤解）

from fastapi import FastAPI

app = FastAPI()

class Service:
    def hello(self) -> str:
        return "hello"

@app.get("/bad")
def bad(service: Service):
    """
    NG：
    service は Depends ではないので FastAPI は生成しない。
    → リクエストのパラメータとして解釈され、422エラーになりがち。
    """
    return {"msg": service.hello()}

上の例では FastAPI は service を「クエリパラメータ等の入力」とみなします。
つまり DIではなく入力扱いになり、リクエスト時にエラーになります。

例：手動で組み立てた場合（DIではなく手動注入）

service = Service()  # アプリ起動時に自分で作る（手動）

@app.get("/manual")
def manual():
    return {"msg": service.hello()}

29-5. 「入力（Request data）」と「DI」の見分け方

FastAPI は引数を以下のどれかとして解釈します：

• パス/クエリ/ヘッダ/ボディ（クライアント入力）
• Depends（サーバ側が用意する依存＝DI）

from fastapi import Header, Body, Depends

def get_token() -> str:
    return "server-generated-token"

@app.post("/mix")
def mix(
    user_id: int,                          # パス/クエリの入力
    x_token: str = Header(...),            # ヘッダ入力
    payload: dict = Body(...),             # ボディ入力
    token: str = Depends(get_token),       # ← DI（サーバ側が注入）
):
    return {"user_id": user_id, "x_token": x_token, "payload": payload, "token": token}

29-6. 詳しい説明（Spring Bootとの違い・スコープ感）

リクエストごとに作られる依存（典型：DBセッション）

def get_db():
    db = DbSession()
    try:
        yield db          # リクエスト中は同じdbを使う
    finally:
        db.close()        # リクエスト終了時に必ず閉じる

アプリ全体で1回だけ作る依存（手動 or キャッシュ）

# 例：設定読み込みなど（概念例）
settings = load_settings_once()

def get_settings():
    return settings

29-7. まとめ

• FastAPIのDIは Dependsを明示した引数だけ注入される
• Dependsが無い引数は 入力（リクエストパラメータ）扱いになる
• 依存は連鎖でき、yield により後始末も安全に書ける
• Spring Bootの「コンテナ管理」より、FastAPIは「関数で明示的に組み立てる」思想

30. SQLite3 を例にしたDBアクセス（Connection / CRUD / Transaction / ORM）

30-1. 概要

Python では標準ライブラリの sqlite3 を使うことで、追加インストールなしで SQLite を扱えます。
配布が簡単（DBファイル1つ）なので、学習・小規模ツール・PCローカルアプリでよく使われます。

30-2. 目的（SQLiteを使う理由）

• サーバ不要でDBが使える（ファイルDB）
• Python標準で使える（sqlite3）
• 小規模なAPIやデスクトップツールの永続化にちょうど良い

30-3. 接続（Connection）とテーブル作成

import sqlite3

DB_PATH = "app.db"

def init_db():
    # DBファイルに接続（無ければ作成される）
    conn = sqlite3.connect(DB_PATH)

    try:
        # SQL実行（テーブル作成）
        conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id   INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            age  INTEGER
        )
        """)
        conn.commit()  # 変更を確定
    finally:
        conn.close()   # 必ず閉じる

init_db()

30-4. CRUD（SQLアクセス：生SQL）

30-4-1. Create（INSERT）

import sqlite3

def create_user(name: str, age: int | None) -> int:
    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()
        # プレースホルダ "?" を必ず使う（SQLインジェクション対策）
        cur.execute("INSERT INTO users(name, age) VALUES(?, ?)", (name, age))
        conn.commit()
        return cur.lastrowid  # AUTOINCREMENTで発行されたID
    finally:
        conn.close()

new_id = create_user("Alice", 20)
print(new_id)

30-4-2. Read（SELECT）

import sqlite3

def find_user_by_id(user_id: int) -> dict | None:
    conn = sqlite3.connect(DB_PATH)
    try:
        # row_factory を使うと結果を辞書っぽく扱える
        conn.row_factory = sqlite3.Row
        cur = conn.cursor()

        cur.execute("SELECT id, name, age FROM users WHERE id = ?", (user_id,))
        row = cur.fetchone()
        if row is None:
            return None

        # sqlite3.Row -> dict
        return dict(row)
    finally:
        conn.close()

print(find_user_by_id(1))

30-4-3. Update（UPDATE）

import sqlite3

def update_user_age(user_id: int, age: int | None) -> int:
    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()
        cur.execute("UPDATE users SET age = ? WHERE id = ?", (age, user_id))
        conn.commit()
        return cur.rowcount  # 更新された行数
    finally:
        conn.close()

print(update_user_age(1, 21))

30-4-4. Delete（DELETE）

import sqlite3

def delete_user(user_id: int) -> int:
    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()
        cur.execute("DELETE FROM users WHERE id = ?", (user_id,))
        conn.commit()
        return cur.rowcount
    finally:
        conn.close()

print(delete_user(1))

30-5. トランザクション（commit / rollback）

トランザクションは「複数SQLをまとめて成功/失敗させる」仕組みです。
途中で例外が起きたら rollback() で取り消します。

import sqlite3

def transfer_example():
    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()

        # ここから同一トランザクション（commitまで）
        cur.execute("INSERT INTO users(name, age) VALUES(?, ?)", ("Bob", 30))
        cur.execute("INSERT INTO users(name, age) VALUES(?, ?)", ("Carol", 25))

        # 例外が無ければ確定
        conn.commit()

    except Exception:
        # 途中で失敗したら全部取り消し
        conn.rollback()
        raise

    finally:
        conn.close()

with（コンテキストマネージャ）で安全に書く

import sqlite3

def create_two_users_with_with():
    # with sqlite3.connect は close を保証する
    with sqlite3.connect(DB_PATH) as conn:
        cur = conn.cursor()
        cur.execute("INSERT INTO users(name, age) VALUES(?, ?)", ("Dave", 40))
        cur.execute("INSERT INTO users(name, age) VALUES(?, ?)", ("Eve", None))
        # with を抜けるとき、例外がなければ commit、あれば rollback される（sqlite3の仕様）

30-6. ORM はあるのか？（ある：SQLAlchemyなど）

SQLite を ORM で扱うなら SQLAlchemy が定番です。
Spring Boot の JPA Entity に近い感覚で、テーブルをクラスとして扱えます。

30-7. ORM版サンプル（SQLAlchemy + SQLite）

30-7-1. モデル定義（テーブル = クラス）

# pip install sqlalchemy
from sqlalchemy import create_engine, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

DB_URL = "sqlite:///app.db"

engine = create_engine(DB_URL, echo=False)  # echo=True でSQLログが見える
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    age: Mapped[int | None] = mapped_column(Integer, nullable=True)

Base.metadata.create_all(engine)  # テーブル作成

30-7-2. CRUD（ORM）

def orm_create_user(name: str, age: int | None) -> int:
    with SessionLocal() as session:
      user = UserModel(name=name, age=age)
      session.add(user)
      session.commit()        # INSERT確定
      session.refresh(user)   # DB採番IDを反映
      return user.id

def orm_find_user(user_id: int) -> UserModel | None:
    with SessionLocal() as session:
      return session.get(UserModel, user_id)

def orm_update_age(user_id: int, age: int | None) -> bool:
    with SessionLocal() as session:
      user = session.get(UserModel, user_id)
      if user is None:
        return False
      user.age = age
      session.commit()
      return True

def orm_delete_user(user_id: int) -> bool:
    with SessionLocal() as session:
      user = session.get(UserModel, user_id)
      if user is None:
        return False
      session.delete(user)
      session.commit()
      return True

30-7-3. トランザクション（ORM）

def orm_transaction_example():
    with SessionLocal() as session:
      try:
        session.add(UserModel(name="T1", age=10))
        session.add(UserModel(name="T2", age=20))
        session.commit()
      except Exception:
        session.rollback()
        raise

30-8. SQLアクセスとORM、どっちを使う？

• 学習・小規模・SQLを理解したい：まず sqlite3（生SQL）
• 中規模以上・保守性：SQLAlchemy（ORM）
• 実務の多く：ORMを使いつつ、難しいクエリだけ生SQLを併用

31. 実務で多い「ORMを基本にしつつ、難しいクエリだけ生SQLを併用」パターン

31-1. 説明（なぜ併用するのか）

実務では、CRUD（単純な登録・更新・削除・単純検索）は ORM（SQLAlchemy）で書くことが多いです。
しかし、次のような「難しいクエリ」は ORM だけだと読みにくくなったり、実装が複雑になりがちです。

• 複雑な JOIN（多段JOIN、LEFT JOIN 連鎖）
• 集計（GROUP BY / HAVING / window関数）
• DB固有機能（SQLiteの拡張、PostgreSQLの特殊構文など）
• パフォーマンス最適化（ヒント、複雑なインデックス活用）
• SQLの方が意図が明確なレポート系クエリ

31-2. サンプル構成（Repositoryで「ORM CRUD」と「Raw SQL集計」を両方持つ）

ここでは SQLite を例に、以下の方針で書きます：

• CRUD（create/find/update/delete） は ORM
• 集計レポート（GROUP BY） は生SQL（Raw SQL）

31-3. サンプルコード（SQLAlchemy + Raw SQL 併用）

31-3-1. ORMモデル定義（テーブル = クラス）

# --- models.py ---
# ORMモデル定義（SQLAlchemy）
from sqlalchemy import Integer, String, ForeignKey, create_engine, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker

DB_URL = "sqlite:///app.db"

# --- DBエンジン（SQLite） ---
engine = create_engine(DB_URL, echo=False)

# --- セッション生成（DB接続の単位） ---
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    """
    UserModel（ORM Entity相当）
    users テーブルに対応
    """
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String, nullable=False)

    orders: Mapped[list["OrderModel"]] = relationship(back_populates="user")

class OrderModel(Base):
    """
    OrderModel（ORM Entity相当）
    orders テーブルに対応
    """
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    amount: Mapped[int] = mapped_column(Integer, nullable=False)

    user: Mapped[UserModel] = relationship(back_populates="orders")

# テーブル作成
Base.metadata.create_all(engine)

31-3-2. Pydantic DTO（APIレスポンス用）

# --- dto.py ---
from pydantic import BaseModel

class UserResponse(BaseModel):
    """
    UserResponse（API出力DTO）
    ORMモデルを直接返さないための型
    """
    id: int
    name: str

class UserOrderSummaryResponse(BaseModel):
    """
    集計結果用DTO（難しいクエリの戻り値）
    userごとの注文数と合計金額
    """
    user_id: int
    user_name: str
    order_count: int
    total_amount: int

31-3-3. Repository（ORM CRUD + Raw SQL集計）

# --- repository.py ---
# Repositoryは「DBアクセスの窓口」
# ここに CRUD（ORM）と 集計（Raw SQL）をまとめると、Service/Controllerがシンプルになる。

from sqlalchemy import text
from models import SessionLocal, UserModel, OrderModel
from dto import UserOrderSummaryResponse

class UserRepository:
    """
    UserRepository（Repository層）
    - 単純CRUDはORMで書く
    - 複雑/集計はRaw SQLを使う
    """

    # --- ORM: Create ---
    def create_user(self, name: str) -> UserModel:
        """ユーザー作成（ORMでINSERT）"""
        with SessionLocal() as session:
            user = UserModel(name=name)
            session.add(user)
            session.commit()
            session.refresh(user)
            return user

    # --- ORM: Create（注文作成） ---
    def create_order(self, user_id: int, amount: int) -> OrderModel:
        """注文作成（ORMでINSERT）"""
        with SessionLocal() as session:
            order = OrderModel(user_id=user_id, amount=amount)
            session.add(order)
            session.commit()
            session.refresh(order)
            return order

    # --- ORM: Read ---
    def find_user(self, user_id: int) -> UserModel | None:
        """ユーザー取得（ORMでSELECT）"""
        with SessionLocal() as session:
            return session.get(UserModel, user_id)

    # --- Raw SQL: 集計（難しいクエリはSQLで書く） ---
    def get_user_order_summary(self) -> list[UserOrderSummaryResponse]:
        """
        userごとの注文数（COUNT）と合計金額（SUM）を返す
        - JOIN + GROUP BY の集計は ORM より SQL の方が読みやすいことが多い
        """
        sql = text("""
        SELECT
            u.id   AS user_id,
            u.name AS user_name,
            COUNT(o.id) AS order_count,
            COALESCE(SUM(o.amount), 0) AS total_amount
        FROM users u
        LEFT JOIN orders o ON o.user_id = u.id
        GROUP BY u.id, u.name
        ORDER BY total_amount DESC
        """)

        with SessionLocal() as session:
            rows = session.execute(sql).mappings().all()
            # mappings() により row["user_id"] のように dict で扱える
            return [
                UserOrderSummaryResponse(
                    user_id=r["user_id"],
                    user_name=r["user_name"],
                    order_count=r["order_count"],
                    total_amount=r["total_amount"],
                )
                for r in rows
            ]

31-3-4. 呼び出し例（Service/Controller側のイメージ）

# --- usage_example.py ---
# ここでは FastAPI ではなく、利用イメージをシンプルに示す

from repository import UserRepository

repo = UserRepository()

# --- ORMでデータ作成 ---
u1 = repo.create_user("Alice")
u2 = repo.create_user("Bob")

repo.create_order(u1.id, 1000)
repo.create_order(u1.id, 500)
repo.create_order(u2.id, 200)

# --- 集計はRaw SQLで取得 ---
summary = repo.get_user_order_summary()
for s in summary:
    print(s.model_dump())

31-4. 詳しい説明（設計の意図）

31-5. まとめ

• 単純CRUDは ORM（SQLAlchemy）が速い・読みやすい
• JOIN/集計/最適化などは 生SQL の方が明確なことが多い
• Repositoryに閉じて「併用」を統制すると保守性が高い

32. ファイルアクセス（小さいテキスト / バイナリ / 大きいテキスト）・アップロード/ダウンロード・テンポラリーファイル

32-1. 概要

Python のファイルI/Oは open() と with が基本です。
小さいファイルは一括読み込み、大きいファイルは逐次読み込み（ストリーム）が定石です。
Web（FastAPI）ではアップロード/ダウンロードが頻出なので、合わせて整理します。

32-2. 小さいテキストファイル（read / write）

設定ファイルや数KB〜数MB程度の小さいテキストなら、一括読み込みで問題ありません。
文字コードは日本語なら utf-8 を明示するのがおすすめです。

読み込み（全部読む）

from pathlib import Path

path = Path("config.txt")

# with を使うと、例外が起きても必ず close される
with path.open("r", encoding="utf-8") as f:
    text = f.read()

print(text)

書き込み（全部書く）

from pathlib import Path

path = Path("output.txt")
content = "Hello\nPython\n"

with path.open("w", encoding="utf-8", newline="\n") as f:
    f.write(content)

32-3. バイナリファイル（画像・PDFなど）

画像やPDFなどは "rb"/"wb" で扱います。
テキストと違い、encoding は使いません。

バイナリコピー（全体が分かる例）

from pathlib import Path

src = Path("image.jpg")
dst = Path("image_copy.jpg")

with src.open("rb") as fin, dst.open("wb") as fout:
    data = fin.read()      # 小さいバイナリなら一括読み込みでもOK
    fout.write(data)

大きめバイナリはチャンクコピー（安全）

from pathlib import Path

def copy_binary_chunked(src_path: str, dst_path: str, chunk_size: int = 1024 * 1024):
    src = Path(src_path)
    dst = Path(dst_path)

    with src.open("rb") as fin, dst.open("wb") as fout:
        while True:
            chunk = fin.read(chunk_size)
            if not chunk:
                break
            fout.write(chunk)

copy_binary_chunked("video.mp4", "video_copy.mp4")

32-4. 大きいテキストファイル（逐次読み込み）

ログ（数百MB〜GB）などは 一括読み込みしないのが鉄則です。
1行ずつ読む、またはチャンクで読む方法を使います。

1行ずつ処理する（最も標準）

from pathlib import Path

log_path = Path("big.log")

error_count = 0

with log_path.open("r", encoding="utf-8", errors="replace") as f:
    for line in f:
        if "ERROR" in line:
            error_count += 1

print("ERROR lines:", error_count)

チャンク読み（巨大ファイル・高速化向け）

from pathlib import Path

def count_keyword_chunked(path: str, keyword: str, chunk_size: int = 1024 * 1024) -> int:
    p = Path(path)
    count = 0

    # 大きいテキストでもバイナリチャンクで読み、最後にデコードする設計もある
    with p.open("rb") as f:
        buffer = b""
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            buffer += chunk
            # 行単位にするため最後の改行までを処理し、残りを次へ
            lines = buffer.split(b"\n")
            buffer = lines.pop()  # 最後の中途半端行
            for bline in lines:
                line = bline.decode("utf-8", errors="replace")
                if keyword in line:
                    count += 1

        # 残り（最後の行）
        if buffer:
            line = buffer.decode("utf-8", errors="replace")
            if keyword in line:
                count += 1

    return count

print(count_keyword_chunked("big.log", "ERROR"))

32-5. FastAPIでのアップロード（UploadFile）

FastAPI ではアップロードは UploadFile を使うのが定石です。
大きいファイルでもメモリに載せずストリーム処理できます。

# --- app_upload.py ---
from fastapi import FastAPI, UploadFile, File
from pathlib import Path
import shutil

app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    """
    アップロードされたファイルを uploads/ に保存する。
    """
    save_path = UPLOAD_DIR / file.filename

    # UploadFile.file は file-like object（バイナリストリーム）
    # shutil.copyfileobj でストリームコピー（メモリ節約）
    with save_path.open("wb") as f:
        shutil.copyfileobj(file.file, f)

    return {"filename": file.filename, "saved_to": str(save_path)}

32-6. FastAPIでのダウンロード（FileResponse / StreamingResponse）

ファイルをそのまま返す（FileResponse）

# --- app_download.py ---
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pathlib import Path

app = FastAPI()
DOWNLOAD_DIR = Path("uploads")

@app.get("/download/{filename}")
def download(filename: str):
    """
    指定ファイルをダウンロードさせる（Content-Disposition付き）。
    """
    path = DOWNLOAD_DIR / filename
    return FileResponse(
        path=str(path),
        filename=filename,  # ダウンロード時のファイル名
        media_type="application/octet-stream",
    )

巨大ファイルをストリーミング（StreamingResponse）

from fastapi.responses import StreamingResponse

def iter_file(path: Path, chunk_size: int = 1024 * 1024):
    with path.open("rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

@app.get("/download-stream/{filename}")
def download_stream(filename: str):
    path = DOWNLOAD_DIR / filename
    return StreamingResponse(
        iter_file(path),
        media_type="application/octet-stream",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )

32-7. テンポラリーファイル（tempfile）作成と使い方

一時ファイルは標準ライブラリ tempfile が安全です。
OSが安全な場所に作り、名前衝突も避けられます。

一時ファイル（処理後に自動削除される）

import tempfile
from pathlib import Path

# delete=True（デフォルト）なら with を抜けると自動削除される
with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8") as tf:
    tf.write("temporary data\n")
    tf.seek(0)
    print(tf.read())
    print("temp path:", tf.name)  # 途中経過でパスが必要な場合に使える

一時ファイルを「残したい」場合（Windowsで便利）

import tempfile
from pathlib import Path

tf = tempfile.NamedTemporaryFile(mode="wb", delete=False)
try:
    tf.write(b"\x00\x01\x02")
    tf.close()
    path = Path(tf.name)
    print("created:", path)
    # ここで別処理に渡す（外部ツールに渡す等）
finally:
    # 実務では最後に削除する（例外時も）
    if Path(tf.name).exists():
        Path(tf.name).unlink()

一時ディレクトリ（複数ファイルの作業に便利）

import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as d:
    workdir = Path(d)
    (workdir / "a.txt").write_text("A", encoding="utf-8")
    (workdir / "b.txt").write_text("B", encoding="utf-8")

    # 作業ディレクトリ内で処理
    for p in workdir.iterdir():
        print(p.name, p.read_text(encoding="utf-8"))

# with を抜けたらディレクトリごと削除される

32-8. まとめ

• 小さいテキスト：read() / write()
• バイナリ：rb / wb、大きいならチャンク
• 大きいテキスト：for line in f（逐次処理）
• アップロード：UploadFile + ストリームコピー
• ダウンロード：FileResponse / StreamingResponse
• テンポラリ：tempfile を使う（安全・衝突回避）

33. FastAPIで「www-form形式（フォーム送信）」のデータは受け取れる？（Collectionで受け取る方法も含む）

33-1. 結論

受け取れます。
FastAPI は JSON だけでなく、HTMLフォームが送る www-form形式（フォーム形式）も受け取れます。
さらに、配列（list）や辞書（dict）のようなCollectionっぽいデータも工夫すれば受け取れます。

33-2. そもそも「www-form形式」とは？（専門用語の解説）

ブラウザのHTMLフォーム（<form>）から送られるデータ形式の代表が2つあります：

• application/x-www-form-urlencoded（一般的なフォーム送信）
  例：name=Alice&age=20
• multipart/form-data（ファイルアップロードを含むフォーム送信）
  例：ファイル＋テキストをまとめて送る

33-3. まず基本：フォームの単項目を受け取る（迷わない最短ルート）

FastAPI でフォームを受け取るには、引数に Form(...) を付けます。
これを書かないとJSON扱いになり、初心者がハマります。

# pip install "fastapi[standard]"
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login")
def login(
    username: str = Form(...),
    password: str = Form(...)
):
    """
    Form(...) を付けると www-form形式 を受け取れる
    """
    return {"username": username, "password_len": len(password)}

省略可能（optional）の例

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/profile")
def profile(
    name: str = Form(...),
    age: int | None = Form(None)  # 省略可能
):
    return {"name": name, "age": age}

33-4. Collection（list）として受け取れる？ → 受け取れる（同じキーを複数回送る）

フォーム形式で配列（list）を送る定番方法は、
同じキーを複数回送ることです。

送信例（フォームのイメージ）：

• tags=python
• tags=fastapi
• tags=sqlite

これを FastAPI 側で list[str] として受け取れます。

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/tags")
def receive_tags(tags: list[str] = Form(...)):
    """
    tags がフォームで複数回送られると list として受け取れる
    """
    return {"count": len(tags), "tags": tags}

33-5. Collection（dict）として受け取れる？ → 原則は工夫が必要

www-form形式は基本が 「キー = 値」の平坦（フラット）構造です。
そのため ネストしたdict（入れ子構造）をそのまま自然に送るのは苦手です。

ただし実務では次のどちらかで解決します：

1. JSON文字列として1項目に入れて送る（おすすめ）
2. キーを工夫して送る（例：user[name] など）

方法A：JSON文字列として送って、サーバでdict化（おすすめ）

送信例：

meta={"role":"admin","level":3}

from fastapi import FastAPI, Form, HTTPException
import json

app = FastAPI()

@app.post("/meta")
def receive_meta(meta: str = Form(...)):
    """
    meta はフォームでは文字列として届くので、json.loads() で dict に変換する
    """
    try:
        meta_dict = json.loads(meta)
        if not isinstance(meta_dict, dict):
            raise ValueError("meta must be JSON object")
        return {"meta": meta_dict}
    except Exception:
        raise HTTPException(status_code=400, detail="meta must be a JSON object string")

方法B：キー名を工夫する（フロントと約束が必要）

例として、以下のように送る設計がありえます：

user_name=Alice
user_age=20

サーバ側でdictにまとめる：

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/user-form")
def receive_user_form(
    user_name: str = Form(...),
    user_age: int = Form(...)
):
    user = {"name": user_name, "age": user_age}
    return {"user": user}

33-6. 「フォーム全部まとめて受け取りたい」場合（FormDataを読む）

「項目が多い」「何が来るか動的」などの場合は、リクエストからフォーム全体を読めます。
これは dictのように扱えるフォームデータです。

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/form-all")
async def form_all(request: Request):
    """
    request.form() でフォーム全体を読む
    - 返り値は FormData（dictっぽい）
    - 同じキーが複数ある場合は getlist() が使える
    """
    form = await request.form()
    data = dict(form)  # 単純なキーは dict化できる

    # 複数値（list）があり得るキーは getlist
    tags = form.getlist("tags")  # tags=... が複数回来るケース

    return {"data": data, "tags": tags}

33-7. まとめ（初心者が迷わない判断基準）

• HTMLフォームから来るなら：Form(...) を使う
• 配列（list）で受けたいなら：同じキーを複数回送る → list[str] = Form(...)
• dict（入れ子）で受けたいなら：フォームは苦手 → JSON文字列で送るのが安全
• 項目が動的なら：await request.form() で全体を読む

34. collection → DB取得 → レスポンス（ページング）までの一括フロー

34-1. 仕様（やりたいこと）

• 一般的な users テーブルからユーザー一覧を取得する
• 一覧画面（表）を想定し、ページング（page / size）に対応する
• 表示仕様：name は「{name}様」として返す（例：Alice → Alice様）
• 返却形式：Collection（list）と、総件数（total）などメタ情報を返す

34-2. 注意点（初心者がハマる所）

34-3. 標準的なフォルダ構成（例）

app/
  main.py                  # FastAPI起動点
  routers/
    users.py               # ルート設定（HTTP入口）
  dto/
    users_dto.py           # Request/Response DTO（Pydantic）
  entities/
    user_entity.py         # 内部モデル（dataclass）
  repositories/
    user_repository.py     # Repository Interface + 実装（DBアクセス）
  services/
    user_service.py        # Service（業務ロジック）
  db/
    sqlite.py              # SQLite接続ユーティリティ

34-4. サンプルコード一式（ページング：users一覧）

34-4-1. DB接続（SQLite）: app/db/sqlite.py

# app/db/sqlite.py
import sqlite3
from typing import Generator

DB_PATH = "app.db"

def get_conn() -> Generator[sqlite3.Connection, None, None]:
    """
    【関数】DBコネクションを提供するDI用関数
    - yield で返す間だけ有効（リクエスト中に使用）
    - 終了時に必ず close される
    """
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 【設定】SELECT結果を dict風に扱える
    try:
        yield conn
    finally:
        conn.close()

34-4-2. Entity（内部モデル）: app/entities/user_entity.py

# app/entities/user_entity.py
from dataclasses import dataclass

@dataclass
class UserEntity:
    """
    【クラス】UserEntity（内部モデル / Entity相当）
    - DBから取った値を内部処理で扱いやすい形にする
    - APIのDTOとは分離する（情報漏えい防止・変更耐性）
    """
    id: int            # 【プロパティ】ユーザーID
    name: str          # 【プロパティ】ユーザー名（素の名前）
    email: str         # 【プロパティ】メール（例として）
    
    def display_name(self) -> str:
        """
        【メソッド】画面表示用の名前を返す
        - 仕様：name + '様'
        """
        return f"{self.name}様"

34-4-3. DTO（Pydantic）: app/dto/users_dto.py

# app/dto/users_dto.py
from pydantic import BaseModel, Field

class UsersPageQuery(BaseModel):
    """
    【クラス】ページング用クエリDTO（Request DTO）
    - page/size をクエリとして受け取る想定
    """
    page: int = Field(1, ge=1, description="ページ番号（1始まり）")     # 【プロパティ】page
    size: int = Field(20, ge=1, le=200, description="1ページ件数")      # 【プロパティ】size

class UserRowResponse(BaseModel):
    """
    【クラス】ユーザー1行のレスポンスDTO（Response DTO）
    """
    id: int                 # 【プロパティ】ユーザーID
    name: str               # 【プロパティ】表示名（name様）
    email: str              # 【プロパティ】メール

class UsersPageResponse(BaseModel):
    """
    【クラス】ユーザー一覧（ページング）レスポンスDTO（Response DTO）
    - items: コレクション（list）
    - total: 総件数
    - page/size: ページ情報
    """
    items: list[UserRowResponse]  # 【プロパティ】ユーザー行のコレクション
    total: int                   # 【プロパティ】総件数
    page: int                    # 【プロパティ】ページ番号（1始まり）
    size: int                    # 【プロパティ】1ページ件数

34-4-4. Repository Interface + 実装（SQLite SQL）: app/repositories/user_repository.py

# app/repositories/user_repository.py
from typing import Protocol
import sqlite3
from app.entities.user_entity import UserEntity

class IUserRepository(Protocol):
    """
    【インターフェース】UserRepositoryの契約（Protocol）
    - 実装差し替え（SQLite ↔ PostgreSQL / FakeRepo）が容易
    """
    def count_all(self) -> int:
        """【メソッド】総件数を返す"""
        ...

    def find_page(self, offset: int, limit: int) -> list[UserEntity]:
        """【メソッド】ページングでユーザー一覧を返す"""
        ...

class SqliteUserRepository:
    """
    【クラス】SQLite実装のRepository
    - DBアクセス（SQL）はこのクラスに閉じ込める
    """
    def __init__(self, conn: sqlite3.Connection):
        """【コンストラクタ】DIされたConnectionを保持する"""
        self.conn = conn  # 【プロパティ】SQLiteコネクション

    def count_all(self) -> int:
        """【メソッド】総件数取得（COUNT）"""
        cur = self.conn.cursor()
        cur.execute("SELECT COUNT(*) AS cnt FROM users")
        row = cur.fetchone()
        return int(row["cnt"])

    def find_page(self, offset: int, limit: int) -> list[UserEntity]:
        """
        【メソッド】ページング取得（LIMIT/OFFSET）
        - offset は 0始まり
        - limit は 1ページ件数
        """
        cur = self.conn.cursor()
        cur.execute(
            "SELECT id, name, email FROM users ORDER BY id LIMIT ? OFFSET ?",
            (limit, offset)
        )
        rows = cur.fetchall()

        # Row -> Entity（内部モデル）に変換して返す
        return [UserEntity(id=r["id"], name=r["name"], email=r["email"]) for r in rows]

34-4-5. Service（業務ロジック）: app/services/user_service.py

# app/services/user_service.py
from app.repositories.user_repository import IUserRepository
from app.dto.users_dto import UsersPageResponse, UserRowResponse

class UserService:
    """
    【クラス】UserService（Service層）
    - ここで業務仕様（name様）を適用する
    - ルートから呼ばれ、Repositoryを利用する
    """
    def __init__(self, repo: IUserRepository):
        """【コンストラクタ】Repositoryを受け取り保持する"""
        self.repo = repo  # 【プロパティ】Repository（Interface）

    def get_users_page(self, page: int, size: int) -> UsersPageResponse:
        """
        【メソッド】ユーザー一覧（ページング）を返す
        - total 件数も返す（表のページングに必要）
        """
        # page(1始まり) -> offset(0始まり) に変換
        offset = (page - 1) * size

        total = self.repo.count_all()
        entities = self.repo.find_page(offset=offset, limit=size)

        # Entity -> DTO（Response）へ変換（表示仕様：name様）
        items = [
            UserRowResponse(
                id=e.id,
                name=e.display_name(),   # ← 仕様：name + '様'
                email=e.email,
            )
            for e in entities
        ]

        return UsersPageResponse(items=items, total=total, page=page, size=size)

34-4-6. DI組み立て関数: app/main.py（依存の生成）

# app/main.py
from fastapi import FastAPI, Depends
import sqlite3

from app.db.sqlite import get_conn
from app.repositories.user_repository import SqliteUserRepository, IUserRepository
from app.services.user_service import UserService
from app.routers.users import router as users_router

app = FastAPI()
app.include_router(users_router)

def get_user_repo(conn: sqlite3.Connection = Depends(get_conn)) -> IUserRepository:
    """
    【関数】RepositoryをDIするための依存
    - ConnectionをDependsで受け取り、Repository実装を返す
    """
    return SqliteUserRepository(conn)

def get_user_service(repo: IUserRepository = Depends(get_user_repo)) -> UserService:
    """
    【関数】ServiceをDIするための依存
    - RepoをDependsで受け取り、Serviceを返す
    """
    return UserService(repo)

34-4-7. ルート設定（Router/Endpoint）: app/routers/users.py

# app/routers/users.py
from fastapi import APIRouter, Depends
from app.dto.users_dto import UsersPageResponse
from app.services.user_service import UserService
from app.main import get_user_service  # ※循環import回避は実務では工夫（後述）

router = APIRouter(prefix="/users", tags=["users"])

@router.get("", response_model=UsersPageResponse)
def list_users(
    page: int = 1,
    size: int = 20,
    service: UserService = Depends(get_user_service),
):
    """
    【ルート】GET /users?page=1&size=20
    - ルートはHTTP入口。DBやSQLを書かない。
    - Serviceに委譲するだけにする。
    """
    return service.get_users_page(page=page, size=size)

34-5. 全体の流れ（キーワード対応）

1. ルート設定（Router）：/users を定義し、page/size を受け取る
2. DI：Depends(get_user_service) で Service を注入
3. Service：ページング計算（offset）＋ Repository 呼び出し
4. Repository：SQLでDB取得（count / page）
5. Entity：DB行を内部モデルへ（UserEntity）
6. DTO：EntityをレスポンスDTOへ変換（name様の仕様を反映）
7. コレクション：items: list[UserRowResponse] を返す

34. collection → DB取得 → レスポンス（ページング）までの一括フロー

34-1. 仕様（やりたいこと）

• 一般的な users テーブルからユーザー一覧を取得する
• 一覧画面（表）を想定し、ページング（page / size）に対応する
• 表示仕様：name は「{name}様」として返す（例：Alice → Alice様）
• 返却形式：Collection（list）と、総件数（total）などメタ情報を返す

34-2. 注意点（初心者がハマる所）

34-3. 標準的なフォルダ構成（例）

app/
  main.py                  # FastAPI起動点
  routers/
    users.py               # ルート設定（HTTP入口）
  dto/
    users_dto.py           # Request/Response DTO（Pydantic）
  entities/
    user_entity.py         # 内部モデル（dataclass）
  repositories/
    user_repository.py     # Repository Interface + 実装（DBアクセス）
  services/
    user_service.py        # Service（業務ロジック）
  db/
    sqlite.py              # SQLite接続ユーティリティ

34-4. サンプルコード一式（ページング：users一覧）

34-4-1. DB接続（SQLite）: app/db/sqlite.py

# app/db/sqlite.py
import sqlite3
from typing import Generator

DB_PATH = "app.db"

def get_conn() -> Generator[sqlite3.Connection, None, None]:
    """
    【関数】DBコネクションを提供するDI用関数
    - yield で返す間だけ有効（リクエスト中に使用）
    - 終了時に必ず close される
    """
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row  # 【設定】SELECT結果を dict風に扱える
    try:
        yield conn
    finally:
        conn.close()

34-4-2. Entity（内部モデル）: app/entities/user_entity.py

# app/entities/user_entity.py
from dataclasses import dataclass

@dataclass
class UserEntity:
    """
    【クラス】UserEntity（内部モデル / Entity相当）
    - DBから取った値を内部処理で扱いやすい形にする
    - APIのDTOとは分離する（情報漏えい防止・変更耐性）
    """
    id: int            # 【プロパティ】ユーザーID
    name: str          # 【プロパティ】ユーザー名（素の名前）
    email: str         # 【プロパティ】メール（例として）
    
    def display_name(self) -> str:
        """
        【メソッド】画面表示用の名前を返す
        - 仕様：name + '様'
        """
        return f"{self.name}様"

34-4-3. DTO（Pydantic）: app/dto/users_dto.py

# app/dto/users_dto.py
from pydantic import BaseModel, Field

class UsersPageQuery(BaseModel):
    """
    【クラス】ページング用クエリDTO（Request DTO）
    - page/size をクエリとして受け取る想定
    """
    page: int = Field(1, ge=1, description="ページ番号（1始まり）")     # 【プロパティ】page
    size: int = Field(20, ge=1, le=200, description="1ページ件数")      # 【プロパティ】size

class UserRowResponse(BaseModel):
    """
    【クラス】ユーザー1行のレスポンスDTO（Response DTO）
    """
    id: int                 # 【プロパティ】ユーザーID
    name: str               # 【プロパティ】表示名（name様）
    email: str              # 【プロパティ】メール

class UsersPageResponse(BaseModel):
    """
    【クラス】ユーザー一覧（ページング）レスポンスDTO（Response DTO）
    - items: コレクション（list）
    - total: 総件数
    - page/size: ページ情報
    """
    items: list[UserRowResponse]  # 【プロパティ】ユーザー行のコレクション
    total: int                   # 【プロパティ】総件数
    page: int                    # 【プロパティ】ページ番号（1始まり）
    size: int                    # 【プロパティ】1ページ件数

34-4-4. Repository Interface + 実装（SQLite SQL）: app/repositories/user_repository.py

# app/repositories/user_repository.py
from typing import Protocol
import sqlite3
from app.entities.user_entity import UserEntity

class IUserRepository(Protocol):
    """
    【インターフェース】UserRepositoryの契約（Protocol）
    - 実装差し替え（SQLite ↔ PostgreSQL / FakeRepo）が容易
    """
    def count_all(self) -> int:
        """【メソッド】総件数を返す"""
        ...

    def find_page(self, offset: int, limit: int) -> list[UserEntity]:
        """【メソッド】ページングでユーザー一覧を返す"""
        ...

class SqliteUserRepository:
    """
    【クラス】SQLite実装のRepository
    - DBアクセス（SQL）はこのクラスに閉じ込める
    """
    def __init__(self, conn: sqlite3.Connection):
        """【コンストラクタ】DIされたConnectionを保持する"""
        self.conn = conn  # 【プロパティ】SQLiteコネクション

    def count_all(self) -> int:
        """【メソッド】総件数取得（COUNT）"""
        cur = self.conn.cursor()
        cur.execute("SELECT COUNT(*) AS cnt FROM users")
        row = cur.fetchone()
        return int(row["cnt"])

    def find_page(self, offset: int, limit: int) -> list[UserEntity]:
        """
        【メソッド】ページング取得（LIMIT/OFFSET）
        - offset は 0始まり
        - limit は 1ページ件数
        """
        cur = self.conn.cursor()
        cur.execute(
            "SELECT id, name, email FROM users ORDER BY id LIMIT ? OFFSET ?",
            (limit, offset)
        )
        rows = cur.fetchall()

        # Row -> Entity（内部モデル）に変換して返す
        return [UserEntity(id=r["id"], name=r["name"], email=r["email"]) for r in rows]

34-4-5. Service（業務ロジック）: app/services/user_service.py

# app/services/user_service.py
from app.repositories.user_repository import IUserRepository
from app.dto.users_dto import UsersPageResponse, UserRowResponse

class UserService:
    """
    【クラス】UserService（Service層）
    - ここで業務仕様（name様）を適用する
    - ルートから呼ばれ、Repositoryを利用する
    """
    def __init__(self, repo: IUserRepository):
        """【コンストラクタ】Repositoryを受け取り保持する"""
        self.repo = repo  # 【プロパティ】Repository（Interface）

    def get_users_page(self, page: int, size: int) -> UsersPageResponse:
        """
        【メソッド】ユーザー一覧（ページング）を返す
        - total 件数も返す（表のページングに必要）
        """
        # page(1始まり) -> offset(0始まり) に変換
        offset = (page - 1) * size

        total = self.repo.count_all()
        entities = self.repo.find_page(offset=offset, limit=size)

        # Entity -> DTO（Response）へ変換（表示仕様：name様）
        items = [
            UserRowResponse(
                id=e.id,
                name=e.display_name(),   # ← 仕様：name + '様'
                email=e.email,
            )
            for e in entities
        ]

        return UsersPageResponse(items=items, total=total, page=page, size=size)

34-4-6. DI組み立て関数: app/main.py（依存の生成）

# app/main.py
from fastapi import FastAPI, Depends
import sqlite3

from app.db.sqlite import get_conn
from app.repositories.user_repository import SqliteUserRepository, IUserRepository
from app.services.user_service import UserService
from app.routers.users import router as users_router

app = FastAPI()
app.include_router(users_router)

def get_user_repo(conn: sqlite3.Connection = Depends(get_conn)) -> IUserRepository:
    """
    【関数】RepositoryをDIするための依存
    - ConnectionをDependsで受け取り、Repository実装を返す
    """
    return SqliteUserRepository(conn)

def get_user_service(repo: IUserRepository = Depends(get_user_repo)) -> UserService:
    """
    【関数】ServiceをDIするための依存
    - RepoをDependsで受け取り、Serviceを返す
    """
    return UserService(repo)

34-4-7. ルート設定（Router/Endpoint）: app/routers/users.py

# app/routers/users.py
from fastapi import APIRouter, Depends
from app.dto.users_dto import UsersPageResponse
from app.services.user_service import UserService
from app.main import get_user_service  # ※循環import回避は実務では工夫（後述）

router = APIRouter(prefix="/users", tags=["users"])

@router.get("", response_model=UsersPageResponse)
def list_users(
    page: int = 1,
    size: int = 20,
    service: UserService = Depends(get_user_service),
):
    """
    【ルート】GET /users?page=1&size=20
    - ルートはHTTP入口。DBやSQLを書かない。
    - Serviceに委譲するだけにする。
    """
    return service.get_users_page(page=page, size=size)

34-5. 全体の流れ（キーワード対応）

1. ルート設定（Router）：/users を定義し、page/size を受け取る
2. DI：Depends(get_user_service) で Service を注入
3. Service：ページング計算（offset）＋ Repository 呼び出し
4. Repository：SQLでDB取得（count / page）
5. Entity：DB行を内部モデルへ（UserEntity）
6. DTO：EntityをレスポンスDTOへ変換（name様の仕様を反映）
7. コレクション：items: list[UserRowResponse] を返す

FastAPIでメッセージ多言語化（i18n）する標準的な方法

1. まず結論（FastAPIに「標準内蔵のi18n」はある？）

FastAPI 自体には Laravel のような「標準で用意された言語ファイル機構」は ありません。
その代わり実務では、次のどれかが「標準的なやり方」として使われます。

1. Accept-Language（HTTPヘッダ）で言語を決める
2. Middleware で「今回のリクエストの言語」を確定して保持する
3. 翻訳辞書（JSON/YAML） または gettext（.po/.mo） を使う
4. エラーは 独自のエラーコード（例：USER_NOT_FOUND）を返し、文言はフロントで翻訳する（APIでは特に多い）

2. 実務で一番おすすめ：APIは「エラーコード＋必要ならメッセージ」

APIはクライアント（Web/Android/iOS）が複数あることが多いため、
文言をAPI側で確定しすぎるより、エラーコードを返す方が保守しやすいです。

# 例：APIレスポンス（言語に依存しない）
{
  "error_code": "USER_NOT_FOUND",
  "message": "User not found"  // optional（開発用に付ける場合も）
}

ただし「管理画面」などAPI利用者が固定で、API側でメッセージを返したい場合は 次の方法が定石です。

3. 定石パターン：Accept-Languageで言語を決めて、翻訳辞書で返す

3-1. 例：簡易辞書（Pythonのdict）で翻訳する（最小構成）

小規模なら、まずはこれで十分です。後でgettext等へ移行もできます。

# app/i18n/messages.py
MESSAGES = {
    "ja": {
        "USER_NOT_FOUND": "ユーザーが見つかりません",
        "WELCOME": "{name}様、ようこそ",
    },
    "en": {
        "USER_NOT_FOUND": "User not found",
        "WELCOME": "Welcome, {name}",
    },
}

def t(lang: str, key: str, **params) -> str:
    """
    【関数】翻訳関数
    - lang: 'ja' / 'en' など
    - key : メッセージキー
    - params: {name} のような埋め込み用
    """
    table = MESSAGES.get(lang) or MESSAGES["en"]       # fallback
    template = table.get(key) or MESSAGES["en"].get(key) or key
    return template.format(**params)

3-2. Middlewareで言語を確定し、request.stateに保存する

request.state は「このリクエスト処理中だけ使える入れ物」です。
Controller/Service からも参照できるようにするのが目的です。

# app/middlewares/locale.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

def pick_lang(accept_language: str | None) -> str:
    """
    【関数】Accept-Languageから言語を決める（簡易版）
    - 'ja,en-US;q=0.9,en;q=0.8' みたいな値が来る
    - ここでは「jaが含まれてたらja、それ以外はen」にする
    """
    if not accept_language:
        return "en"
    lower = accept_language.lower()
    if "ja" in lower:
        return "ja"
    return "en"

class LocaleMiddleware(BaseHTTPMiddleware):
    """
    【クラス】言語決定ミドルウェア
    - ルートに入る前に lang を決めて request.state.lang に入れる
    """
    async def dispatch(self, request: Request, call_next):
        accept = request.headers.get("accept-language")
        request.state.lang = pick_lang(accept)
        response = await call_next(request)
        return response

3-3. FastAPIにミドルウェアを登録する

# app/main.py
from fastapi import FastAPI
from app.middlewares.locale import LocaleMiddleware

app = FastAPI()
app.add_middleware(LocaleMiddleware)

3-4. Controllerで翻訳して返す例（業務例外→翻訳→HTTPレスポンス）

# app/controllers/users_controller.py
from fastapi import Request, HTTPException
from app.i18n.messages import t

class UsersController:
    """
    【クラス】UsersController
    - 例外をHTTPに変換するときに、langに合わせたmessageを返す
    """

    def raise_not_found(self, request: Request):
        """
        【メソッド】ユーザー未存在エラーを多言語で返す例
        """
        lang = getattr(request.state, "lang", "en")
        msg = t(lang, "USER_NOT_FOUND")
        raise HTTPException(status_code=404, detail=msg)

    def welcome(self, request: Request, name: str) -> dict:
        """
        【メソッド】表示メッセージを多言語で返す例
        """
        lang = getattr(request.state, "lang", "en")
        msg = t(lang, "WELCOME", name=name)
        return {"message": msg}

4. 本格運用（gettext / Babel）を使う場合

文言が増える・翻訳者が関わる・差分管理が必要、という場合は gettext（.po/.mo） が定番です。
ただし導入が少し重いので、最初は「辞書方式」で十分 → 必要になったら移行、が現実的です。

5. 実務のおすすめ指針（迷わない結論）

• APIが複数クライアント向けなら：エラーコード返却＋文言はフロント翻訳が最強
• APIが管理画面など限定的なら：Accept-Language + Middleware + 翻訳関数が定石
• 最初は辞書方式でOK、必要になったらgettextに移行

Python / FastAPI のログ出力：デバッグ用（コンソール）と本番用（ログファイル）の使い分け

1. まず結論（使い分けの基本）

2. ログレベル（重要：何をどこまで出すか）

3. 開発用：コンソールに出す（最小構成）

開発時は コンソール出力（StreamHandler） が基本です。
VS Code のターミナル / デバッグコンソールで見えます。

# app/logging_config.py
import logging

def setup_logging_dev() -> None:
    """
    【関数】開発用ロギング設定
    - コンソールに DEBUG 以上を出す
    - まずはこれでOK
    """
    logging.basicConfig(
        level=logging.DEBUG,  # 【設定】DEBUG以上を表示
        format="%(asctime)s %(levelname)s %(name)s - %(message)s",
    )

# app/main.py
import logging
from fastapi import FastAPI
from app.logging_config import setup_logging_dev

setup_logging_dev()

logger = logging.getLogger("app")  # 【変数】アプリ用logger
app = FastAPI()

@app.get("/ping")
def ping():
    """
    【ルート】動作確認
    """
    logger.debug("debug: ping called")   # 【ログ】デバッグ
    logger.info("info: ping ok")         # 【ログ】通常情報
    return {"ok": True}

4. 本番用：ログファイルに残す（ローテーション付き）

本番はログが増え続けるため、ローテーション（一定サイズや日付で分割）が必須です。

# app/logging_config.py
import logging
from logging.handlers import RotatingFileHandler
import os

def setup_logging_prod(log_dir: str = "logs") -> None:
    """
    【関数】本番用ロギング設定
    - ログファイルに INFO 以上を出す
    - ローテーションあり（サイズで分割）
    """
    os.makedirs(log_dir, exist_ok=True)  # 【処理】ログフォルダ作成
    log_path = os.path.join(log_dir, "app.log")

    handler = RotatingFileHandler(
        filename=log_path,         # 【設定】ログファイル
        maxBytes=5 * 1024 * 1024,  # 【設定】5MBで分割
        backupCount=10,            # 【設定】古いログを10世代保持
        encoding="utf-8",          # 【設定】文字化け防止
    )

    formatter = logging.Formatter(
        "%(asctime)s %(levelname)s %(name)s - %(message)s"
    )
    handler.setFormatter(formatter)

    root = logging.getLogger()     # 【取得】root logger
    root.setLevel(logging.INFO)    # 【設定】INFO以上
    root.handlers.clear()          # 【注意】二重
::contentReference[oaicite:0]{index=0}

# app/main.py  
import logging
from fastapi import FastAPI
from app.logging_config import setup_logging_prod
setup_logging_prod()  # 【呼び出し】本番用ロギング設定
logger = logging.getLogger("app")  # 【変数】アプリ用logger
app = FastAPI()
@app.get("/ping")
def ping():
    """
    【ルート】動作確認
    """
    logger.debug("debug: ping called")   # 【ログ】デバッグ（本番では出ない）
    logger.info("info: ping ok")         # 【ログ】通常情報
    return {"ok": True}

Python + FastAPI で「Windowsのiniファイル」的な設定はどこに置き、いつ、どう読み込む？（サンプル付き）

1. まず結論（実務での標準パターン）

2. 推奨フォルダ構成（iniを置く場所）

app/
  main.py
  dependencies.py
  ...
config/
  app.ini          # アプリ設定（DB接続、外部API、機能フラグなど）
  app.dev.ini      # 開発用（任意）
  app.prod.ini     # 本番用（任意）

3. iniファイル例（config/app.ini）

[app]
env = dev
debug = true

[db]
driver = sqlite
path = app.db

[external_api]
base_url = https://example.com
timeout_seconds = 10

4. 読み込みコード（標準：configparser）

4-1. 設定読み込みモジュールを作る（1回読み＋キャッシュ）

# app/config_loader.py
from __future__ import annotations
import os
from pathlib import Path
import configparser
from functools import lru_cache

@lru_cache(maxsize=1)
def load_config() -> configparser.ConfigParser:
    """
    【関数】ini設定を読み込んで返す（キャッシュされる）
    - maxsize=1 なので最初の1回だけファイルを読む
    - 以降はメモリ上のConfigParserが返る（高速）
    """
    # 【環境変数】iniのパスを切り替えられるようにする（本番で便利）
    ini_path = os.getenv("APP_INI_PATH", "config/app.ini")

    path = Path(ini_path)
    if not path.exists():
        raise FileNotFoundError(f"ini file not found: {path}")

    config = configparser.ConfigParser()
    config.read(path, encoding="utf-8")  # 【処理】iniを読む（UTF-8推奨）
    return config

5. 「いつ読む？」の標準的な2パターン

パターンA：import時に読む（最も簡単）

小規模アプリならこれで十分です。
ただし import 時に例外が出ると起動できないので、運用では好みが分かれます。

# app/settings.py
from app.config_loader import load_config

CONFIG = load_config()  # 【処理】起動時（import時）に1回読む

APP_ENV = CONFIG.get("app", "env", fallback="dev")
DB_PATH = CONFIG.get("db", "path", fallback="app.db")

パターンB：startupイベントで読む（FastAPIらしい）

FastAPI起動時に確実に読ませたいならこちら。
startupで読み込みが失敗したらアプリ起動を止められます。

# app/main.py
import logging
from fastapi import FastAPI
from app.config_loader import load_config

logger = logging.getLogger("app")
app = FastAPI()

@app.on_event("startup")
def on_startup():
    """
    【イベント】アプリ起動時に1回呼ばれる
    - ここでiniが読めることを確認する
    """
    cfg = load_config()
    env = cfg.get("app", "env", fallback="dev")
    logger.info("config loaded. env=%s", env)

6. FastAPIで「必要な場所に設定を渡す」（DIで注入）

SpringのDIに近い形で、Controller/Serviceに設定を渡せます。

# app/dependencies.py
import configparser
from fastapi import Depends
from app.config_loader import load_config

def get_config() -> configparser.ConfigParser:
    """
    【DI関数】設定を返す
    - load_config() はキャッシュされるので高速
    """
    return load_config()

# app/routers/ping.py
from fastapi import APIRouter, Depends
import configparser
from app.dependencies import get_config

router = APIRouter()

@router.get("/ping")
def ping(cfg: configparser.ConfigParser = Depends(get_config)):
    """
    【ルート】ini設定を参照する例
    """
    env = cfg.get("app", "env", fallback="dev")
    timeout = cfg.getint("external_api", "timeout_seconds", fallback=10)
    return {"ok": True, "env": env, "timeout_seconds": timeout}

7. 本番での切り替え（iniファイルのパスを変える）

本番では ini の場所が変わることが多いので、
環境変数でiniパスを渡すのが定石です。

# Windows（PowerShell例）
$env:APP_INI_PATH="C:\app\config\app.prod.ini"
uvicorn app.main:app --host 0.0.0.0 --port 8000

8. さらに実務っぽい注意（iniを使う場合）

• 秘密情報（DBパスワード/APIキー）は ini ではなく、可能なら 環境変数に寄せる
• iniは「起動時に1回読む」前提（リクエスト毎に読むのは避ける）
• 複数プロセス起動（uvicorn workers）でも、各プロセスが起動時に読む（普通）

秘密情報は ini ではなく環境変数へ：設定方法（Windows/Linux）・読み込み方・いつ読むか

1. なぜ秘密情報は環境変数に寄せるのか（実務理由）

2. Windowsでの環境変数の設定方法

2-1. PowerShell（このターミナルだけ有効：一時）

# DBパスワード / APIキー例
$env:DB_PASSWORD = "secret_password"
$env:API_KEY     = "secret_api_key"

# 起動
uvicorn app.main:app --reload

2-2. CMD（このウィンドウだけ有効：一時）

set DB_PASSWORD=secret_password
set API_KEY=secret_api_key

uvicorn app.main:app --reload

2-3. Windowsの「システム環境変数」に登録（永続）

1. スタート → 「環境変数」と検索 → 「システム環境変数を編集」
2. 「環境変数(N)...」
3. 「ユーザー環境変数」または「システム環境変数」で追加
4. 追加後、新しく開いたターミナルから有効

3. Linuxでの環境変数の設定方法

3-1. シェルで一時設定（このセッションだけ）

export DB_PASSWORD="secret_password"
export API_KEY="secret_api_key"

uvicorn app.main:app --host 0.0.0.0 --port 8000

3-2. .bashrc / .zshrc に書く（ユーザー永続）

# ~/.bashrc 例
export DB_PASSWORD="secret_password"
export API_KEY="secret_api_key"

# 反映（例）
source ~/.bashrc

3-3. systemdサービスで設定（本番の定石）

本番でFastAPIをsystemdで常駐させる場合は、Unitファイルに環境変数を渡します。

# /etc/systemd/system/myapp.service（例）
[Service]
Environment="DB_PASSWORD=secret_password"
Environment="API_KEY=secret_api_key"
ExecStart=/usr/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000

# /etc/systemd/system/myapp.service（例：EnvironmentFile）
[Service]
EnvironmentFile=/etc/myapp/myapp.env
ExecStart=/usr/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000

# /etc/myapp/myapp.env（例）
DB_PASSWORD=secret_password
API_KEY=secret_api_key

4. Pythonでの読み込み方（os.getenv が基本）

Pythonでは標準で os.getenv() や os.environ を使います。

# app/secrets.py
import os

def get_db_password() -> str:
    """
    【関数】DBパスワードを環境変数から取得する
    - 無い場合は起動できないので例外にする（秘密はデフォルト値を持たない）
    """
    value = os.getenv("DB_PASSWORD")
    if not value:
        raise RuntimeError("DB_PASSWORD is not set")
    return value

def get_api_key() -> str:
    """
    【関数】APIキーを環境変数から取得する
    """
    value = os.getenv("API_KEY")
    if not value:
        raise RuntimeError("API_KEY is not set")
    return value

5. いつ読むのが正しいか（タイミングの標準）

5-1. 起動時に1回読む（おすすめ）

起動時に「必須の秘密情報が揃っているか」検証して、無ければ起動を止めるのが安全です。
Webアプリは「途中で設定が変わる」運用は基本しないため、起動時読みが定石です。

# app/main.py
import logging
from fastapi import FastAPI
from app.secrets import get_db_password, get_api_key

logger = logging.getLogger("app")
app = FastAPI()

@app.on_event("startup")
def on_startup():
    """
    【イベント】アプリ起動時に1回呼ばれる
    - 必須の秘密情報が揃っているか検証する
    """
    _db_pass = get_db_password()
    _api_key = get_api_key()
    logger.info("secrets loaded (lengths only): db_password=%s api_key=%s",
                len(_db_pass), len(_api_key))

5-2. 必要になった時に読む（DIで注入する方式）

「必要な箇所でだけ使いたい」「テストで差し替えたい」場合はDIにします。
ただし実務では、内部でキャッシュして「毎回環境変数を読む」ようにはしません。

# app/dependencies.py
from fastapi import Depends
from functools import lru_cache
from app.secrets import get_api_key

@lru_cache(maxsize=1)
def load_api_key_once() -> str:
    """
    【関数】APIキーを1回だけ読み込んでキャッシュする
    """
    return get_api_key()

def get_api_key_dep() -> str:
    """
    【DI関数】APIキーを注入する
    - 実体はキャッシュされているので高速
    """
    return load_api_key_once()

# app/routers/external.py
from fastapi import APIRouter, Depends
from app.dependencies import get_api_key_dep

router = APIRouter()

@router.get("/call-external")
def call_external(api_key: str = Depends(get_api_key_dep)):
    """
    【ルート】外部API呼び出し（例）
    """
    # ここで api_key を使って外部APIを叩く（例）
    return {"ok": True}

6. ini と環境変数を混ぜる定石（iniは非秘密、秘密はenv）

実務では以下の分離が分かりやすいです。

• iniに置く：DBホスト、DB名、タイムアウト、機能フラグなど（漏れても致命的でない）
• 環境変数に置く：DBパスワード、APIキー、JWT秘密鍵など（漏れると終わる）

# 例：DB接続情報の組み立て（疑似例）
db_user = ini_cfg.get("db", "user")
db_pass = os.getenv("DB_PASSWORD")  # 秘密はenv
dsn = f"postgresql://{db_user}:{db_pass}@{host}:{port}/{db_name}"

7. よくある落とし穴（初心者向け）

• VS Codeターミナルを開き直さないと、GUIで設定した環境変数が反映されない
• APIキーをログに出す（絶対NG）
• 設定漏れに気づかない：秘密情報にデフォルト値を付けてしまう
• 複数プロセス（workers）では 各プロセスが起動時に読む（普通）

ユーザーテーブルにユーザーを追加する（FastAPI + ORM / 業務レベル例外処理）

1. 仕様（やりたいこと）

• users テーブルにユーザーを追加（INSERT）する
• カラム：id, name, tel, email, isactive, updated_at
• API：POST /users
• リクエストボディ：JSON
• 必須：name, tel, email
• tel / email は形式チェック。エラーなら 400 Bad Request
• email が既に登録済みなら 400 を返し、エラー情報をJSONで返す
• DBアクセスは ORM（SQLAlchemy）を使用
• INSERT時にDBエラーが起きたら 409 Conflict（Bodyに詳細コード）
• INSERT時、isactive は 1(true) に設定

2. 注意事項（初心者がハマるポイント）

3. フォルダ構成（例）

app/
  main.py
  dependencies.py
  routers/
    users_routes.py
  controllers/
    users_controller.py
  services/
    users_service.py
  repositories/
    users_repository.py
  db/
    session.py
    models.py
  dto/
    users_dto.py
  responses/
    error_response.py
  errors/
    app_errors.py

4. コード一式（Router / Controller / Service / Repository / DB / DTO / Response）

4-1. エラー型（業務例外）: app/errors/app_errors.py

# app/errors/app_errors.py

class BusinessError(Exception):
    """
    【例外】業務例外の基底
    - ControllerがHTTPエラーへ変換する対象
    """
    def __init__(self, error_code: str, message: str, details: dict | None = None):
        super().__init__(message)
        self.error_code = error_code          # 【プロパティ】業務エラーコード
        self.message = message                # 【プロパティ】ユーザー向けメッセージ（必要なら）
        self.details = details or {}          # 【プロパティ】追加情報（フィールド名など）

class RequestValidationError(BusinessError):
    """
    【例外】入力値の業務的な不正（400）
    """
    pass

class DuplicateEmailError(BusinessError):
    """
    【例外】email重複（400）
    """
    pass

class DbInsertConflictError(BusinessError):
    """
    【例外】DBのINSERT競合/制約違反など（409）
    """
    pass

4-2. エラーレスポンスDTO: app/responses/error_response.py

# app/responses/error_response.py
from pydantic import BaseModel

class ErrorResponse(BaseModel):
    """
    【DTO】エラーレスポンス統一形式
    - error_code: 機械判定用
    - message: 人間向け
    - details: フィールド名や原因など追加情報
    """
    error_code: str
    message: str
    details: dict

4-3. DTO（Request/Response）: app/dto/users_dto.py

# app/dto/users_dto.py
from pydantic import BaseModel, Field, EmailStr, constr

# tel形式（例）：数字とハイフンのみ、10〜13文字程度（例：090-1234-5678 / 0312345678）
TelStr = constr(pattern=r"^[0-9-]{10,13}$")

class CreateUserRequest(BaseModel):
    """
    【DTO】ユーザー作成リクエスト（POST Body）
    - name/tel/email 必須
    - tel/email 形式チェックあり
    """
    name: str = Field(..., min_length=1, max_length=100, description="ユーザー名（必須）")
    tel: TelStr = Field(..., description="電話番号（必須：数字とハイフン、10〜13文字）")
    email: EmailStr = Field(..., description="メールアドレス（必須：形式チェックあり）")

class CreateUserResponse(BaseModel):
    """
    【DTO】ユーザー作成レスポンス
    """
    id: int
    name: str
    tel: str
    email: str
    isactive: bool
    updated_at: str

4-4. DBセッション（SQLAlchemy）: app/db/session.py

# app/db/session.py
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

# SQLite例（実務では環境変数で切り替え）
DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLiteのスレッド制約回避（FastAPI用）
    pool_pre_ping=True,                         # 接続死活チェック
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator[Session, None, None]:
    """
    【DI関数】DBセッションを提供する
    - リクエスト中は同じSessionを使う
    - 最後に必ずcloseする
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4-5. ORMモデル: app/db/models.py

# app/db/models.py
from datetime import datetime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, Boolean, DateTime, UniqueConstraint

class Base(DeclarativeBase):
    """
    【ベース】SQLAlchemy Declarative Base
    """
    pass

class UserModel(Base):
    """
    【ORM】usersテーブル
    - emailはUNIQUE（重複登録禁止）
    - updated_at は INSERT/UPDATEで自動更新
    """
    __tablename__ = "users"
    __table_args__ = (
        UniqueConstraint("email", name="uq_users_email"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)  # 【PK】id
    name: Mapped[str] = mapped_column(String(100), nullable=False)                   # 【必須】name
    tel: Mapped[str] = mapped_column(String(20), nullable=False)                     # 【必須】tel
    email: Mapped[str] = mapped_column(String(255), nullable=False)                  # 【必須】email（UNIQUE）
    isactive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)    # 【必須】isactive
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow
    )

4-6. DI（Repository/Serviceを組み立て）: app/dependencies.py

# app/dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.repositories.users_repository import UsersRepository
from app.services.users_service import UsersService

def get_users_repository(db: Session = Depends(get_db)) -> UsersRepository:
    """
    【DI関数】UsersRepositoryを生成する
    """
    return UsersRepository(db)

def get_users_service(repo: UsersRepository = Depends(get_users_repository)) -> UsersService:
    """
    【DI関数】UsersServiceを生成する
    """
    return UsersService(repo)

4-7. Repository（ORMアクセス）: app/repositories/users_repository.py

# app/repositories/users_repository.py
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

from app.db.models import UserModel

class UsersRepository:
    """
    【Repository】users テーブルアクセス（ORM）
    - SQLAlchemy Sessionを使ってDB操作を行う
    """

    def __init__(self, db: Session):
        """
        【コンストラクタ】DBセッションを受け取って保持する
        """
        self.db = db  # 【プロパティ】SQLAlchemy Session

    def exists_by_email(self, email: str) -> bool:
        """
        【メソッド】emailが既に存在するか確認する
        """
        stmt = select(UserModel.id).where(UserModel.email == email)
        row = self.db.execute(stmt).first()
        return row is not None

    def insert_user(self, name: str, tel: str, email: str) -> UserModel:
        """
        【メソッド】ユーザーをINSERTして、作成されたUserModelを返す
        - isactive は true を強制
        - commit/rollbackはここで行う（業務レベルの責務分担として）
        """
        user = UserModel(
            name=name,
            tel=tel,
            email=email,
            isactive=True,     # 【仕様】insert時は必ず true
        )

        try:
            self.db.add(user)        # 【処理】INSERT対象として登録
            self.db.commit()         # 【処理】トランザクション確定
            self.db.refresh(user)    # 【処理】DB採番IDなどを反映
            return user              # 【返却】作成されたユーザー

        except IntegrityError:
            # 【例外】UNIQUE制約違反など（email重複を含む可能性）
            self.db.rollback()
            raise

        except SQLAlchemyError:
            # 【例外】その他DB例外
            self.db.rollback()
            raise

4-8. Service（業務ロジック＋業務例外）: app/services/users_service.py

# app/services/users_service.py
from sqlalchemy.exc import IntegrityError, SQLAlchemyError

from app.repositories.users_repository import UsersRepository
from app.errors.app_errors import DuplicateEmailError, DbInsertConflictError

class UsersService:
    """
    【Service】ユーザー作成の業務ロジック
    - email重複チェック
    - RepositoryでINSERT
    - DB例外を業務例外へ変換
    """

    def __init__(self, repo: UsersRepository):
        """
        【コンストラクタ】Repositoryを受け取り保持する
        """
        self.repo = repo  # 【プロパティ】UsersRepository

    def create_user(self, name: str, tel: str, email: str):
        """
        【メソッド】ユーザーを作成する
        - email重複なら 400
        - INSERT失敗なら 409
        """

        # 【業務】email重複チェック（分かりやすい事前チェック）
        if self.repo.exists_by_email(email):
            raise DuplicateEmailError(
                error_code="DUPLICATE_EMAIL",
                message="email is already registered",
                details={"field": "email", "email": email},
            )

        # 【DB】INSERT（Repositoryがcommitまで実施）
        try:
            return self.repo.insert_user(name=name, tel=tel, email=email)

        except IntegrityError:
            # 【例外】同時実行での競合などでUNIQUE違反が起きた場合もここに来る
            # → 要件では「email登録済みは400」なので、ここも400として扱う
            raise DuplicateEmailError(
                error_code="DUPLICATE_EMAIL",
                message="email is already registered",
                details={"field": "email", "email": email, "reason": "unique_constraint"},
            )

        except SQLAlchemyError as e:
            # 【例外】INSERT時のDBエラー → 409 Conflict（詳細コードを付ける）
            raise DbInsertConflictError(
                error_code="DB_INSERT_FAILED",
                message="failed to insert user",
                details={"reason": str(e.__class__.__name__)},
            )

4-9. Controller（HTTP入口：例外→HTTP変換 / エラーJSON統一）: app/controllers/users_controller.py

# app/controllers/users_controller.py
import logging
from fastapi import HTTPException

from app.services.users_service import UsersService
from app.dto.users_dto import CreateUserRequest, CreateUserResponse
from app.responses.error_response import ErrorResponse
from app.errors.app_errors import DuplicateEmailError, DbInsertConflictError

logger = logging.getLogger("app")

class UsersController:
    """
    【Controller】HTTP入口の処理
    - Serviceを呼ぶ
    - 業務例外をHTTPステータスに変換
    - エラーボディは ErrorResponse 形式で返す
    """

    def create_user(self, req: CreateUserRequest, service: UsersService) -> CreateUserResponse:
        """
        【メソッド】POST /users の処理
        """
        try:
            user = service.create_user(name=req.name, tel=req.tel, email=req.email)

            # 【返却】レスポンスDTO（updated_atはISO文字列にして返す）
            return CreateUserResponse(
                id=user.id,
                name=user.name,
                tel=user.tel,
                email=user.email,
                isactive=bool(user.isactive),
                updated_at=user.updated_at.isoformat(),
            )

        except DuplicateEmailError as e:
            # 【業務】email重複（要件：400 + JSONでエラー情報）
            body = ErrorResponse(error_code=e.error_code, message=e.message, details=e.details).model_dump()
            raise HTTPException(status_code=400, detail=body)

        except DbInsertConflictError as e:
            # 【業務】INSERT失敗（要件：409 + 詳細コード）
            body = ErrorResponse(error_code=e.error_code, message=e.message, details=e.details).model_dump()
            raise HTTPException(status_code=409, detail=body)

        except Exception as e:
            # 【想定外】業務レベルではログを残して500
            logger.exception("unexpected error in create_user")
            body = ErrorResponse(
                error_code="INTERNAL_ERROR",
                message="internal server error",
                details={"reason": str(e.__class__.__name__)},
            ).model_dump()
            raise HTTPException(status_code=500, detail=body)

4-10. Router（URL紐付け＋DI注入）: app/routers/users_routes.py

# app/routers/users_routes.py
from fastapi import APIRouter, Depends
from app.dto.users_dto import CreateUserRequest, CreateUserResponse
from app.services.users_service import UsersService
from app.dependencies import get_users_service
from app.controllers.users_controller import UsersController

router = APIRouter(prefix="/users", tags=["users"])
controller = UsersController()

@router.post("", response_model=CreateUserResponse)
def create_user(
    body: CreateUserRequest,                                   # 【引数】JSON Body（DTOで形式チェック）
    service: UsersService = Depends(get_users_service),         # 【DI】Service注入
) -> CreateUserResponse:
    """
    【ルート】POST /users
    - RouterはURL紐付けのみ
    - 実処理はControllerへ委譲
    """
    return controller.create_user(req=body, service=service)

4-11. main.py（起動点＋テーブル作成例）: app/main.py

# app/main.py
import logging
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError as FastApiRequestValidationError

from app.routers.users_routes import router as users_router
from app.db.session import engine
from app.db.models import Base
from app.responses.error_response import ErrorResponse

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s")
logger = logging.getLogger("app")

app = FastAPI()
app.include_router(users_router)

@app.on_event("startup")
def on_startup():
    """
    【イベント】起動時処理
    - 例：テーブル作成（学習用）
    - 実務では migration ツールを使うことが多い
    """
    Base.metadata.create_all(bind=engine)
    logger.info("app started")

@app.exception_handler(FastApiRequestValidationError)
async def validation_exception_handler(request, exc):
    """
    【例外ハンドラ】Pydantic/入力形式エラーを 400 に統一する
    - FastAPIのデフォルトは422になりがちなので、要件に合わせて400へ寄せる
    """
    body = ErrorResponse(
        error_code="BAD_REQUEST",
        message="invalid request",
        details={"errors": exc.errors()},
    ).model_dump()
    return JSONResponse(status_code=400, content={"detail": body})

5. 返却されるエラーJSONの例

5-1. email重複（400）

{
  "detail": {
    "error_code": "DUPLICATE_EMAIL",
    "message": "email is already registered",
    "details": {
      "field": "email",
      "email": "alice@example.com"
    }
  }
}

5-2. INSERT失敗（409）

{
  "detail": {
    "error_code": "DB_INSERT_FAILED",
    "message": "failed to insert user",
    "details": {
      "reason": "OperationalError"
    }
  }
}

注文・請求ルールエンジン（FastAPI版：業務ロジック中心）

1. 仕様（やりたいこと）

入力（注文JSON）を受け取り、税・割引・送料を計算して、請求結果を返します。
「HTTP通信」よりも、業務ロジックを安全に実装することを主目的とします。

1-1. 入力（注文JSON）

{
  "customerRank": "GOLD",
  "region": "TOKYO",
  "items": [
    {"sku": "A001", "unitPrice": 980,  "qty": 2, "tax": "STANDARD"},
    {"sku": "B777", "unitPrice": 1500, "qty": 1, "tax": "REDUCED"}
  ],
  "coupon": "WELCOME10"
}

1-2. 出力（計算結果）

{
  "subtotal": 3460,
  "taxTotal": 292,
  "discount": 346,
  "shipping": 500,
  "grandTotal": 3906
}

1-3. 業務ルール

• 税（TaxCategory）
  • STANDARD：10%
  • REDUCED：8%
  • EXEMPT：0%
• 割引（CustomerRank / Coupon）
  • GOLD会員：小計の 5%
  • WELCOME10クーポン：小計の 10%
  • 割引は 会員割 + クーポン割 を合算
• 送料（Region / FreeShipping）
  • TOKYO：500円
  • OTHER：800円
  • 小計が 10,000円以上で送料無料（送料=0）

2. 注意点（業務として安全に書くコツ）

3. コード（Router / Controller / Service / DTO / Enum / Response）

3-1. フォルダ構成（例）

app/
  main.py
  routers/
    billing_routes.py
  controllers/
    billing_controller.py
  services/
    billing_rule_engine.py
  dto/
    billing_dto.py
  enums/
    billing_enums.py
  responses/
    error_response.py

3-2. Enum定義: app/enums/billing_enums.py

# app/enums/billing_enums.py
from enum import Enum

class CustomerRank(str, Enum):
    """
    【Enum】顧客ランク
    - 入力値を固定し、タイポや想定外値を弾く
    """
    BRONZE = "BRONZE"
    SILVER = "SILVER"
    GOLD = "GOLD"

class Region(str, Enum):
    """
    【Enum】配送地域
    """
    TOKYO = "TOKYO"
    OTHER = "OTHER"

class TaxCategory(str, Enum):
    """
    【Enum】税区分
    """
    STANDARD = "STANDARD"
    REDUCED = "REDUCED"
    EXEMPT = "EXEMPT"

class CouponCode(str, Enum):
    """
    【Enum】クーポン（任意）
    - 必要最小限の例
    """
    WELCOME10 = "WELCOME10"

3-3. DTO（入力/出力）: app/dto/billing_dto.py

# app/dto/billing_dto.py
from pydantic import BaseModel, Field
from typing import Optional, List

from app.enums.billing_enums import CustomerRank, Region, TaxCategory, CouponCode

class OrderItemDto(BaseModel):
    """
    【DTO】注文行
    """
    sku: str = Field(..., min_length=1, max_length=50, description="商品コード")
    unitPrice: int = Field(..., ge=0, description="単価（円）")
    qty: int = Field(..., ge=1, description="数量")
    tax: TaxCategory = Field(..., description="税区分（Enum）")

class OrderInputDto(BaseModel):
    """
    【DTO】注文入力（Request Body）
    - customerRank / region / tax はEnumとして受け取る
    - coupon は optional（無い場合は None）
    """
    customerRank: CustomerRank = Field(..., description="顧客ランク")
    region: Region = Field(..., description="配送地域")
    items: List[OrderItemDto] = Field(..., min_length=1, description="注文行配列")
    coupon: Optional[CouponCode] = Field(None, description="クーポン（任意）")

class BillingResultDto(BaseModel):
    """
    【DTO】計算結果（Response）
    """
    subtotal: int
    taxTotal: int
    discount: int
    shipping: int
    grandTotal: int

3-4. エラーレスポンス（業務向けの形）: app/responses/error_response.py

# app/responses/error_response.py
from pydantic import BaseModel

class ErrorResponse(BaseModel):
    """
    【DTO】エラーレスポンスの統一形式
    """
    error_code: str
    message: str
    details: dict

3-5. Service（ルールエンジン本体）: app/services/billing_rule_engine.py

# app/services/billing_rule_engine.py
from dataclasses import dataclass
from typing import Tuple

from app.dto.billing_dto import OrderInputDto, BillingResultDto
from app.enums.billing_enums import TaxCategory, CustomerRank, Region, CouponCode

@dataclass(frozen=True)
class BillingRules:
    """
    【定数】業務ルール値（マジックナンバー排除）
    """
    TAX_STANDARD: float = 0.10
    TAX_REDUCED: float = 0.08
    TAX_EXEMPT: float = 0.00

    GOLD_DISCOUNT: float = 0.05
    WELCOME10_DISCOUNT: float = 0.10

    SHIPPING_TOKYO: int = 500
    SHIPPING_OTHER: int = 800
    FREE_SHIPPING_THRESHOLD: int = 10000

class BillingRuleEngine:
    """
    【Service】請求計算ルールエンジン
    - 税/割引/送料/合計の計算本体をここに集約
    """

    def __init__(self, rules: BillingRules = BillingRules()):
        """
        【コンストラクタ】ルール値を注入（テスト差し替え可能）
        """
        self.rules = rules

    def calc_tax_rate(self, tax: TaxCategory) -> float:
        """
        【メソッド】税区分（Enum）から税率を返す
        """
        # matchはPython 3.10+（Windowsでも普通に使える）
        match tax:
            case TaxCategory.STANDARD:
                return self.rules.TAX_STANDARD
            case TaxCategory.REDUCED:
                return self.rules.TAX_REDUCED
            case TaxCategory.EXEMPT:
                return self.rules.TAX_EXEMPT
        # Enumなので原則ここには来ない（保険）
        return self.rules.TAX_EXEMPT

    def calc_member_discount_rate(self, rank: CustomerRank) -> float:
        """
        【メソッド】会員ランクから割引率を返す
        """
        if rank == CustomerRank.GOLD:
            return self.rules.GOLD_DISCOUNT
        return 0.0

    def calc_coupon_discount_rate(self, coupon: CouponCode | None) -> float:
        """
        【メソッド】クーポンから割引率を返す（任意）
        """
        if coupon == CouponCode.WELCOME10:
            return self.rules.WELCOME10_DISCOUNT
        return 0.0

    def calc_shipping(self, region: Region, subtotal: int) -> int:
        """
        【メソッド】送料を計算する（送料無料条件あり）
        """
        if subtotal >= self.rules.FREE_SHIPPING_THRESHOLD:
            return 0

        match region:
            case Region.TOKYO:
                return self.rules.SHIPPING_TOKYO
            case Region.OTHER:
                return self.rules.SHIPPING_OTHER
        return self.rules.SHIPPING_OTHER

    def calc(self, order: OrderInputDto) -> BillingResultDto:
        """
        【メソッド】注文入力から請求結果を計算して返す（業務の中心）
        - 行処理：forで安全に回す
        - 金額：int（円）
        """
        # 1) 行小計・税額を集計
        subtotal = 0
        tax_total = 0

        for item in order.items:
            line_subtotal = item.unitPrice * item.qty
            subtotal += line_subtotal

            rate = self.calc_tax_rate(item.tax)
            # 金額は int（円）なので丸めが必要。ここでは四捨五入（業務要件次第）
            line_tax = int(round(line_subtotal * rate))
            tax_total += line_tax

        # 2) 割引（会員 + クーポン）を合算
        member_rate = self.calc_member_discount_rate(order.customerRank)
        coupon_rate = self.calc_coupon_discount_rate(order.coupon)
        discount_rate = member_rate + coupon_rate

        discount = int(round(subtotal * discount_rate))

        # 3) 送料
        shipping = self.calc_shipping(order.region, subtotal)

        # 4) 合計
        grand_total = subtotal + tax_total - discount + shipping

        return BillingResultDto(
            subtotal=subtotal,
            taxTotal=tax_total,
            discount=discount,
            shipping=shipping,
            grandTotal=grand_total,
        )

3-6. Controller（HTTP入口は薄く）: app/controllers/billing_controller.py

# app/controllers/billing_controller.py
import logging
from fastapi import HTTPException

from app.dto.billing_dto import OrderInputDto, BillingResultDto
from app.responses.error_response import ErrorResponse
from app.services.billing_rule_engine import BillingRuleEngine

logger = logging.getLogger("app")

class BillingController:
    """
    【Controller】HTTP受け口（薄く）
    - 入力DTOはFastAPI/Pydanticが検証済み
    - ここではServiceを呼ぶだけ
    - 想定外はログを残して500
    """

    def __init__(self, engine: BillingRuleEngine):
        """
        【コンストラクタ】RuleEngineを受け取る
        """
        self.engine = engine

    def calculate(self, order: OrderInputDto) -> BillingResultDto:
        """
        【メソッド】請求計算を実行する
        """
        try:
            return self.engine.calc(order)
        except Exception as e:
            logger.exception("billing calculation failed")
            body = ErrorResponse(
                error_code="BILLING_CALC_FAILED",
                message="billing calculation failed",
                details={"reason": e.__class__.__name__},
            ).model_dump()
            raise HTTPException(status_code=500, detail=body)

3-7. Router（DI組み立て）: app/routers/billing_routes.py

# app/routers/billing_routes.py
from fastapi import APIRouter, Depends
from app.dto.billing_dto import OrderInputDto, BillingResultDto
from app.services.billing_rule_engine import BillingRuleEngine
from app.controllers.billing_controller import BillingController

router = APIRouter(prefix="/billing", tags=["billing"])

def get_billing_engine() -> BillingRuleEngine:
    """
    【DI関数】RuleEngineを生成して返す
    - ルールが固定ならシングルトンでもよい
    """
    return BillingRuleEngine()

def get_billing_controller(engine: BillingRuleEngine = Depends(get_billing_engine)) -> BillingController:
    """
    【DI関数】Controllerを生成して返す
    """
    return BillingController(engine)

@router.post("/calculate", response_model=BillingResultDto)
def calculate(
    order: OrderInputDto,
    controller: BillingController = Depends(get_billing_controller),
) -> BillingResultDto:
    """
    【ルート】POST /billing/calculate
    - RouterはHTTP入口（薄く）
    - DTO検証後にControllerへ委譲する
    """
    return controller.calculate(order)

3-8. main.py（起動点）: app/main.py

# app/main.py
import logging
from fastapi import FastAPI
from app.routers.billing_routes import router as billing_router

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s")

app = FastAPI()
app.include_router(billing_router)

4. 動作確認用サンプル（curl）

curl -X POST http://127.0.0.1:8000/billing/calculate ^
  -H "Content-Type: application/json" ^
  -d "{\"customerRank\":\"GOLD\",\"region\":\"TOKYO\",\"items\":[{\"sku\":\"A001\",\"unitPrice\":980,\"qty\":2,\"tax\":\"STANDARD\"},{\"sku\":\"B777\",\"unitPrice\":1500,\"qty\":1,\"tax\":\"REDUCED\"}],\"coupon\":\"WELCOME10\"}"

FastAPI + SQLite3：DBファイル生成 → テーブル作成 → 初期データ投入（起動時/セットアップ時）

1. 結論：FastAPIでも可能。ただし「HTTPのたびに作る」はやらない

2. 推奨フロー（Mermaid）

flowchart TD
    A[アプリ起動 / テスト開始] --> B[DBファイルの存在チェック]
    B -->|無い| C[SQLiteファイル生成（接続すれば自動作成）]
    B -->|ある| D[次へ]
    C --> E[テーブル作成（Base.metadata.create_all）]
    D --> E[テーブル作成（必要に応じて）]
    E --> F[初期データ投入（seed）]
    F --> G[API起動完了]

3. フォルダ構成（例：起動時initルーチンを分離）

app/
  main.py
  db/
    session.py
    models.py
    init_db.py          # ★ここに「作成＆seed」を集約
  routers/
    ...

4. コード例（SQLAlchemy + SQLite）

4-1. DB接続（SQLite）: app/db/session.py

# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

4-2. ORMモデル（例：users）: app/db/models.py

# app/db/models.py
from datetime import datetime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String, Boolean, DateTime, UniqueConstraint

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    __tablename__ = "users"
    __table_args__ = (UniqueConstraint("email", name="uq_users_email"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    tel: Mapped[str] = mapped_column(String(20), nullable=False)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    isactive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow
    )

4-3. DB初期化（テーブル作成＋seed）: app/db/init_db.py

# app/db/init_db.py
from sqlalchemy.orm import Session
from sqlalchemy import select
from app.db.session import engine, SessionLocal
from app.db.models import Base, UserModel

def create_tables() -> None:
    """
    【関数】テーブル作成（ローカル/テスト用）
    - SQLiteなら、engine接続でDBファイルが自動生成される
    - create_all は「無ければ作る」だが、HTTPのたびに呼ばない
    """
    Base.metadata.create_all(bind=engine)

def seed_initial_data(db: Session) -> None:
    """
    【関数】初期データ投入（Seeder相当）
    - 例：管理者ユーザー1件を入れる
    - 既に存在する場合は二重投入しない
    """
    admin_email = "admin@example.com"

    # 既存チェック（INSERT重複防止）
    exists = db.execute(select(UserModel.id).where(UserModel.email == admin_email)).first()
    if exists:
        return

    admin = UserModel(
        name="Admin",
        tel="000-0000-0000",
        email=admin_email,
        isactive=True,
    )
    db.add(admin)
    db.commit()

def init_db_for_local_or_test() -> None:
    """
    【関数】初期化ルーチン（まとめ）
    - テーブル作成
    - seed投入
    """
    create_tables()
    db = SessionLocal()
    try:
        seed_initial_data(db)
    finally:
        db.close()

4-4. 起動時に一度だけ呼ぶ（FastAPI startup）: app/main.py

# app/main.py
import os
import logging
from fastapi import FastAPI

from app.db.init_db import init_db_for_local_or_test

logger = logging.getLogger("app")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s")

app = FastAPI()

@app.on_event("startup")
def on_startup():
    """
    【イベント】起動時に1回だけ呼ばれる
    - ローカル/テスト環境だけ DB初期化する（本番で勝手に作らない）
    """
    env = os.getenv("APP_ENV", "dev")
    if env in ("dev", "test"):
        init_db_for_local_or_test()
        logger.info("db initialized for %s", env)
    else:
        logger.info("skip db init (env=%s)", env)

5. 本番ではどうする？（業務の定石）

6. 使い方（例）

# Windows PowerShell（開発）
$env:APP_ENV="dev"
uvicorn app.main:app --reload

# Linux（開発）
export APP_ENV=dev
uvicorn app.main:app --reload

Alembic migration（Laravelのmigration相当）を最小構成で導入する手順（FastAPI + SQLAlchemy + SQLite例）

1. 仕様（この手順で実現すること）

• DBスキーマ変更を コードで管理できるようにする（履歴が残る）
• Laravelの migration と同じく「差分を順番に適用」できるようにする
• 開発/テスト/本番で同じ手順（upgrade）でテーブルを作れる
• （任意）SQLAlchemyモデルから migration を 自動生成（autogenerate）できる

2. 注意点（業務の常識）

3. フォルダ構成（最小例）

app/
  main.py
  db/
    session.py
    models.py          # SQLAlchemy ORMモデル（Base含む）
alembic.ini
alembic/
  env.py
  script.py.mako
  versions/
    xxxxx_initial_users.py

4. 導入手順（コマンド）

4-1. インストール

# 例：同期SQLAlchemy + Alembic + SQLite
pip install sqlalchemy alembic

4-2. Alembic初期化（プロジェクト直下で）

alembic init alembic

これで alembic/ フォルダと alembic.ini が生成されます。

5. コード（最小構成）

5-1. DB URLを定義（SQLAlchemy engine）: app/db/session.py

# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./app.db"  # 本番は環境変数に寄せるのが定石

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},  # SQLite + FastAPIの定番設定
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

5-2. ORMモデル定義（Base含む）: app/db/models.py

# app/db/models.py
from datetime import datetime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String, Boolean, DateTime, UniqueConstraint

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    __tablename__ = "users"
    __table_args__ = (UniqueConstraint("email", name="uq_users_email"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    tel: Mapped[str] = mapped_column(String(20), nullable=False)
    email: Mapped[str] = mapped_column(String(255), nullable=False)
    isactive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow
    )

6. Alembic設定（ここが肝）

6-1. alembic.ini のDB URLを設定

生成された alembic.ini を開き、以下を設定します。

; alembic.ini
sqlalchemy.url = sqlite:///./app.db

6-2. alembic/env.py に「モデルのBase」を教える（autogenerate用）

生成された alembic/env.py を編集して、 target_metadata にSQLAlchemyの Base.metadata を設定します。

# alembic/env.py（重要部分だけ抜粋・最小例）
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context

import os
import sys

# app を import できるようにパス追加（最小構成の定番）
sys.path.append(os.path.abspath("."))

from app.db.models import Base  # ← ここが重要（モデルのBase）
# もし engine を使いたいなら：from app.db.session import DATABASE_URL

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# ★ autogenerateが参照するメタデータ
target_metadata = Base.metadata


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,  # 型変更検出をある程度助ける
        )

        with context.begin_transaction():
            context.run_migrations()


run_migrations_online()

7. 初回 migration を作る（initial）

7-1. autogenerateで作成

alembic revision --autogenerate -m "initial users"

alembic/versions/xxxx_initial_users.py が生成されます。

7-2. 生成されたmigration（例）

# alembic/versions/xxxx_initial_users.py（例：概略）
from alembic import op
import sqlalchemy as sa

revision = "xxxx"
down_revision = None

def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
        sa.Column("name", sa.String(length=100), nullable=False),
        sa.Column("tel", sa.String(length=20), nullable=False),
        sa.Column("email", sa.String(length=255), nullable=False),
        sa.Column("isactive", sa.Boolean(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), nullable=False),
    )
    op.create_unique_constraint("uq_users_email", "users", ["email"])

def downgrade() -> None:
    op.drop_table("users")

8. migration を適用（テーブル作成）

# 最新まで適用（Laravelの migrate 相当）
alembic upgrade head

これで app.db が無ければ作られ、テーブルが作成されます。

8-1. 状態確認コマンド

# 現在適用されている revision を表示
alembic current

# 履歴表示
alembic history --verbose

9. どう運用するのが定石か（開発/本番）

9-1. 開発（ローカル）

# モデル変更
# ↓
alembic revision --autogenerate -m "add xxx"
# ↓
alembic upgrade head
# ↓
アプリ起動（uvicorn）

9-2. 本番

10. （任意）DB URLを環境変数にする（本番向け）

alembic.ini を固定にせず、環境変数で切り替える場合は、 alembic/env.py で sqlalchemy.url を上書きします。

# alembic/env.py（冒頭あたりで追加）
db_url = os.getenv("DATABASE_URL")
if db_url:
    config.set_main_option("sqlalchemy.url", db_url)

# Windows PowerShell
$env:DATABASE_URL="sqlite:///./app.db"
alembic upgrade head

# Linux
export DATABASE_URL="sqlite:///./app.db"
alembic upgrade head

UserSetting（JSON）をログイン時に読み込み、全処理から参照できる設定クラスに保持する（FastAPI版）

1. 仕様（FastAPIに置き換え）

• user_settings テーブルは user_id と settings(JSON) を持つ
• ログイン成功時にDBから settings JSON を読み込み、Session に保存する
• 各リクエスト開始時に Session → UserSettingsStore へ復元し、全処理から参照できるようにする
• settings が存在しない場合は デフォルト設定（例：ページング 20）を使う
• DB例外はログに詳細を残し、APIには 安全なエラー（業務レベル）を返す

2. 注意点（FastAPI/Starletteの前提）

3. フォルダ構成（例）

app/
  main.py
  middleware/
    user_settings_middleware.py
  support/
    user_settings_store.py
  db/
    session.py
    models.py
  repositories/
    user_settings_repository.py
  services/
    auth_service.py
  controllers/
    auth_controller.py
  dto/
    auth_dto.py
    error_response.py
  routers/
    auth_routes.py
    sample_routes.py

4. コード一式（Router / Controller / Service / Repository / DB / DTO / Middleware / Store）

4-1. UserSettingsStore（全層から参照するためのクラス）: app/support/user_settings_store.py

# app/support/user_settings_store.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Any

def _get_by_path(obj: dict, path: str) -> Any:
    """
    【関数】"paging.per_page" のようなドット区切りパスで dict を辿る
    - 無ければ KeyError を投げる
    """
    cur: Any = obj
    for part in path.split("."):
        if not isinstance(cur, dict) or part not in cur:
            raise KeyError(path)
        cur = cur[part]
    return cur

@dataclass
class UserSettingsStore:
    """
    【クラス】ユーザー設定の参照ストア（リクエスト内で有効）
    - settings: dict（JSONを読み込んだもの）
    """
    settings: dict

    def get(self, path: str, default: Any = None) -> Any:
        """
        【メソッド】任意型で取得
        """
        try:
            return _get_by_path(self.settings, path)
        except KeyError:
            return default

    def get_int(self, path: str, default: int) -> int:
        """
        【メソッド】intで取得（型が違う/無いならdefault）
        """
        value = self.get(path, default)
        try:
            return int(value)
        except Exception:
            return default

4-2. DB（SQLAlchemy + SQLite）: app/db/session.py

# app/db/session.py
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False},
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator[Session, None, None]:
    """
    【DI関数】DBセッション（リクエスト単位）
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4-3. ORMモデル（user_settings）: app/db/models.py

# app/db/models.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, Text

class Base(DeclarativeBase):
    pass

class UserSettingsModel(Base):
    """
    【ORM】user_settings
    - SQLiteでは JSON 型が環境で揺れるため、最小構成では Text に JSON文字列を保存する例にする
    """
    __tablename__ = "user_settings"

    user_id: Mapped[int] = mapped_column(Integer, primary_key=True)  # user_id PK
    settings: Mapped[str] = mapped_column(Text, nullable=False)      # JSON文字列（必須）

4-4. Repository（settings取得）: app/repositories/user_settings_repository.py

# app/repositories/user_settings_repository.py
import json
import logging
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError

from app.db.models import UserSettingsModel

logger = logging.getLogger("app")

class UserSettingsRepository:
    """
    【Repository】user_settings 取得
    """

    def __init__(self, db: Session):
        """
        【コンストラクタ】DBセッションを受け取る
        """
        self.db = db

    def find_settings_dict_by_user_id(self, user_id: int) -> dict | None:
        """
        【メソッド】user_id の settings(JSON) を dict にして返す
        - 無ければ None
        - DB例外は上位で業務エラー化するため raise する
        """
        try:
            stmt = select(UserSettingsModel.settings).where(UserSettingsModel.user_id == user_id)
            row = self.db.execute(stmt).first()
            if not row:
                return None

            settings_json = row[0]
            return json.loads(settings_json)

        except SQLAlchemyError:
            logger.exception("DB error while loading user_settings (user_id=%s)", user_id)
            raise

        except json.JSONDecodeError:
            # 業務としては「壊れた設定」なので、安全にデフォルトへ落とす等の方針もある
            logger.exception("Invalid JSON in user_settings (user_id=%s)", user_id)
            raise

4-5. DTO（ログイン入力/出力、エラー）: app/dto/auth_dto.py / app/dto/error_response.py

# app/dto/auth_dto.py
from pydantic import BaseModel, Field

class LoginRequest(BaseModel):
    """
    【DTO】ログイン入力
    """
    username: str = Field(..., min_length=1)
    password: str = Field(..., min_length=1)

class LoginResponse(BaseModel):
    """
    【DTO】ログイン成功応答（例）
    """
    ok: bool
    user_id: int

# app/dto/error_response.py
from pydantic import BaseModel

class ErrorResponse(BaseModel):
    """
    【DTO】業務エラー応答
    """
    error_code: str
    message: str
    details: dict

4-6. Service（ログイン成功時に settings を読み込んで session に保存）: app/services/auth_service.py

# app/services/auth_service.py
import logging
from sqlalchemy.exc import SQLAlchemyError

from app.repositories.user_settings_repository import UserSettingsRepository

logger = logging.getLogger("app")

DEFAULT_SETTINGS: dict = {
    "paging": {"per_page": 20},
    "theme": "light",
}

class AuthService:
    """
    【Service】認証＋ユーザー設定ロード
    - 例では認証は簡略化（実務ではDB/IdP等）
    """

    def __init__(self, settings_repo: UserSettingsRepository):
        """
        【コンストラクタ】Repository注入
        """
        self.settings_repo = settings_repo

    def authenticate(self, username: str, password: str) -> int:
        """
        【メソッド】認証（例）
        - 実務ではユーザーテーブル照会、ハッシュ照合など
        - ここでは説明のため固定で user_id を返す
        """
        if username == "demo" and password == "demo":
            return 1
        return 0  # 0は認証失敗扱い

    def load_user_settings_or_default(self, user_id: int) -> dict:
        """
        【メソッド】user_settings を読み込む。無ければデフォルト
        - DB例外は上に投げて、Controllerで安全なエラーへ変換する
        """
        try:
            settings = self.settings_repo.find_settings_dict_by_user_id(user_id)
            if settings is None:
                return DEFAULT_SETTINGS
            return settings
        except SQLAlchemyError:
            logger.exception("Failed to load user_settings (user_id=%s)", user_id)
            raise

4-7. Controller（業務レベル例外処理＋Session保存）: app/controllers/auth_controller.py

# app/controllers/auth_controller.py
import logging
from fastapi import HTTPException, Request
from sqlalchemy.exc import SQLAlchemyError

from app.dto.auth_dto import LoginRequest, LoginResponse
from app.dto.error_response import ErrorResponse
from app.services.auth_service import AuthService

logger = logging.getLogger("app")

class AuthController:
    """
    【Controller】ログイン処理
    - 成功時：settingsをDBから読み→session保存
    - 失敗時：安全なエラーを返す
    """

    def __init__(self, auth_service: AuthService):
        """
        【コンストラクタ】Service注入
        """
        self.auth_service = auth_service

    def login(self, request: Request, body: LoginRequest) -> LoginResponse:
        """
        【メソッド】POST /auth/login
        """
        user_id = self.auth_service.authenticate(body.username, body.password)
        if user_id == 0:
            err = ErrorResponse(
                error_code="AUTH_FAILED",
                message="invalid username or password",
                details={},
            ).model_dump()
            raise HTTPException(status_code=401, detail=err)

        # ログイン成功：settingsを読み込む（無ければデフォルト）
        try:
            settings = self.auth_service.load_user_settings_or_default(user_id)
        except SQLAlchemyError:
            # DB詳細はログに、APIは安全に
            logger.exception("Login succeeded but settings load failed (user_id=%s)", user_id)
            err = ErrorResponse(
                error_code="SETTINGS_LOAD_FAILED",
                message="failed to load user settings",
                details={},
            ).model_dump()
            raise HTTPException(status_code=500, detail=err)

        # Sessionへ保存（次リクエスト以降も維持）
        # ※ Starlette SessionsMiddleware により request.session が使える
        request.session["user_id"] = user_id
        request.session["user_settings"] = settings

        return LoginResponse(ok=True, user_id=user_id)

4-8. Middleware（各リクエスト開始時：session → store へ復元）: app/middleware/user_settings_middleware.py

# app/middleware/user_settings_middleware.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

from app.support.user_settings_store import UserSettingsStore
from app.services.auth_service import DEFAULT_SETTINGS

class UserSettingsRestoreMiddleware(BaseHTTPMiddleware):
    """
    【Middleware】各リクエスト開始時に session から UserSettingsStore を復元する
    - request.state に載せることで、Controller/Serviceから参照できる
    """

    async def dispatch(self, request: Request, call_next) -> Response:
        settings = request.session.get("user_settings") or DEFAULT_SETTINGS
        request.state.user_settings_store = UserSettingsStore(settings=settings)
        return await call_next(request)

4-9. 依存性（どこからでも参照できるようにする）: app/main.py 内 or app/dependencies.py

# app/main.py（dependenciesもここに最小実装する例）
from fastapi import Depends, Request
from app.support.user_settings_store import UserSettingsStore

def get_user_settings_store(request: Request) -> UserSettingsStore:
    """
    【DI関数】UserSettingsStoreを取得する
    - request.state から取り出す（Middlewareで復元済み）
    """
    return request.state.user_settings_store

4-10. Router（ログイン）: app/routers/auth_routes.py

# app/routers/auth_routes.py
from fastapi import APIRouter, Depends, Request
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.repositories.user_settings_repository import UserSettingsRepository
from app.services.auth_service import AuthService
from app.controllers.auth_controller import AuthController
from app.dto.auth_dto import LoginRequest, LoginResponse

router = APIRouter(prefix="/auth", tags=["auth"])

def get_auth_controller(db: Session = Depends(get_db)) -> AuthController:
    """
    【DI関数】Controller生成（Repo→Service→Controller）
    """
    repo = UserSettingsRepository(db)
    service = AuthService(repo)
    return AuthController(service)

@router.post("/login", response_model=LoginResponse)
def login(
    request: Request,
    body: LoginRequest,
    controller: AuthController = Depends(get_auth_controller),
) -> LoginResponse:
    """
    【ルート】POST /auth/login
    """
    return controller.login(request=request, body=body)

4-11. 「全処理から参照」例（Controller/ServiceどこでもOK）: app/routers/sample_routes.py

# app/routers/sample_routes.py
from fastapi import APIRouter, Depends
from app.support.user_settings_store import UserSettingsStore
from app.main import get_user_settings_store

router = APIRouter(prefix="/sample", tags=["sample"])

@router.get("/paging")
def get_paging_settings(store: UserSettingsStore = Depends(get_user_settings_store)):
    """
    【ルート】どの層でも参照できる形のデモ
    """
    per_page = store.get_int("paging.per_page", 20)
    theme = store.get("theme", "light")
    return {"per_page": per_page, "theme": theme}

4-12. main.py（SessionMiddleware + RestoreMiddleware を組み込み）: app/main.py

# app/main.py
import logging
from fastapi import FastAPI
from starlette.middleware.sessions import SessionsMiddleware

from app.db.session import engine
from app.db.models import Base
from app.middleware.user_settings_middleware import UserSettingsRestoreMiddleware
from app.routers.auth_routes import router as auth_router
from app.routers.sample_routes import router as sample_router

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s")

app = FastAPI()

# 【Session】署名付きCookieセッション（最小構成）
# ※ 実務では SECRET_KEY は環境変数から読み込む
app.add_middleware(SessionsMiddleware, secret_key="CHANGE_ME_TO_SECURE_RANDOM")

# 【Restore】各リクエスト開始時に session → store 復元
app.add_middleware(UserSettingsRestoreMiddleware)

# ルータ登録
app.include_router(auth_router)
app.include_router(sample_router)

@app.on_event("startup")
def on_startup():
    """
    【起動時】テーブル作成（学習用）
    - 本番ではAlembic migration推奨
    """
    Base.metadata.create_all(bind=engine)

5. 動作イメージ（実行順）

6. Laravelの完成形に相当する「参照方法」（FastAPI版）

# 例：どこでも（Router/Controller/Service）で
per_page = store.get_int("paging.per_page", 20)
theme    = store.get("theme", "light")

FastAPI + Python：/hello で Hello, World を返す（VS Code / venv / 1ファイル）

1) venv環境の作り方（Windows / VS Code ターミナル）

1. 作業フォルダを作って VS Code で開く（例：C:\work\fastapi-hello）
2. VS Code のターミナルを開く（Terminal → New Terminal）
3. venv を作成：

cd C:\work
mkdir fastapi-hello
cd fastapi-hello

python -m venv .venv

venv を有効化

# PowerShell の場合
.\.venv\Scripts\Activate.ps1

# cmd の場合
.\.venv\Scripts\activate.bat

成功するとプロンプト先頭に (.venv) が付きます。:contentReference[oaicite:0]{index=0}

2) FastAPI のインストール方法

python -m pip install --upgrade pip
pip install fastapi "uvicorn[standard]"

※ uvicorn は FastAPI アプリを起動するサーバー役です。:contentReference[oaicite:1]{index=1}

3) 1 pythonファイルでの実装（main.py）

作業フォルダ直下に main.py を作成して、以下を貼り付け：

    
from __future__ import annotations

import asyncio
import json
from pathlib import Path
from typing import Any, Dict, List

from fastapi import FastAPI, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse

app = FastAPI(
    title="Fetch Debug Mock API",
    version="2.0.0",
    description="Frontend fetch() debugging mock server (advanced)",
)

# =============================================================================
# CORS設定（開発用・完全許可モード）
# -----------------------------------------------------------------------------
# 目的:
#   - フロント側がどのホスト/どのポート/どのフレームワークで動いていても、
#     ほぼ追加設定なしで fetch() デバッグができるようにする。
#   - 例: React(3000), Vite(5173), Vue, Angular, 素HTML(80), 別PC(LAN), Docker など。
#
# 何が解決できるか:
#   - ブラウザのCORSブロックを回避し、"呼べているか/レスポンスが何か" を即確認できる。
#
# 注意（重要）:
#   - allow_origins=["*"] は「全オリジン許可」。
#   - 仕様上 allow_credentials=True（Cookie送信許可）とは併用できない。
#   - Cookie/セッション認証の検証が必要なら、allow_origins を具体的なOriginに絞ること。
#   - 本設定は「ローカル開発・モック用途専用」。本番では使用しない。
# =============================================================================
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # どこからでもOK（開発用）
    allow_methods=["*"],  # GET/POST/PUT/DELETE/OPTIONS... 全部OK
    allow_headers=["*"],  # Content-Type, Authorization など任意ヘッダOK
)

# =============================================================================
# ローカルJSONファイルパス（main.py と同じフォルダに mock.json を置く）
# -----------------------------------------------------------------------------
# 目的:
#   - UI開発のための固定モックデータを、DBや外部API無しで返す。
#   - バックエンド未完成でもフロントを先に作れるようにする。
#
# 使い方:
#   - main.py と同じフォルダに mock.json を置く。
# =============================================================================
MOCK_JSON_FILE = Path(__file__).with_name("mock.json")


# =============================================================================
# GET /hello
# -----------------------------------------------------------------------------
# 目的:
#   - fetch() の最小疎通確認（textレスポンス確認）。
#   - まず「リクエストが届くか」「CORSが通るか」「text() で読めるか」を確認する。
#
# 入力:
#   - なし
#
# 出力:
#   - 200 OK
#   - Content-Type: text/plain
#   - Body: "Hello, World"
#
# フロント側の確認例:
#   fetch("http://127.0.0.1:8000/hello").then(r => r.text()).then(console.log);
# =============================================================================
@app.get("/hello", response_class=PlainTextResponse)
def hello() -> str:
    return "Hello, World"


# =============================================================================
# GET /hellojson
# -----------------------------------------------------------------------------
# 目的:
#   - fetch().then(r => r.json()) の動作確認。
#   - JSONレスポンスの最小形で、JSONパースとUI表示の流れを確認する。
#
# 入力:
#   - なし
#
# 出力:
#   - 200 OK
#   - Content-Type: application/json
#   - Body: {"message":"hello, world"}
#
# フロント側の確認例:
#   fetch("http://127.0.0.1:8000/hellojson").then(r => r.json()).then(console.log);
# =============================================================================
@app.get("/hellojson", response_class=JSONResponse)
def hello_json() -> Dict[str, str]:
    return {"message": "hello, world"}


# =============================================================================
# GET /helloerror
# -----------------------------------------------------------------------------
# 目的:
#   - フロント側のエラー処理をテストする。
#   - response.ok === false, status=500 の分岐、UI通知（トースト/ダイアログ）等を確認する。
#
# 入力:
#   - なし
#
# 出力:
#   - 500 Internal Server Error
#   - Content-Type: application/json
#   - Body: {"error":"..."}
#
# フロント側の確認例:
#   fetch("/helloerror").then(r => console.log(r.ok, r.status));
# =============================================================================
@app.get("/helloerror")
def hello_error() -> JSONResponse:
    return JSONResponse(
        status_code=500,
        content={"error": "Intentional server error for fetch debugging"},
    )


# =============================================================================
# GET /hellojsonfile
# -----------------------------------------------------------------------------
# 目的:
#   - 固定モックデータ(JSONファイル)を返すことで、UI開発をバックエンドから分離する。
#   - 「実データっぽい形」をファイルで管理でき、UI開発・デバッグが容易になる。
#
# 使用方法:
#   - main.py と同じフォルダに mock.json を置く。
#
# 正常時:
#   - 200 OK
#   - Content-Type: application/json
#   - Body: mock.json の内容をそのまま返却
#
# 異常時:
#   - mock.json が無い → 500 + ヒント
#   - mock.json が壊れている（JSONとして不正）→ 500 + どこが壊れているか
#
# 補足:
#   - ここでいう「解析しない」は「業務的意味づけの検証をしない」ことで、
#     JSONとしてロード（パース）し、その結果を返すこと自体は行う。
# =============================================================================
@app.get("/hellojsonfile")
def hello_json_file() -> JSONResponse:
    try:
        if not MOCK_JSON_FILE.exists():
            return JSONResponse(
                status_code=500,
                content={
                    "error": "mock.json not found",
                    "hint": f"Create {MOCK_JSON_FILE.name} next to main.py",
                },
            )

        data: Any = json.loads(MOCK_JSON_FILE.read_text(encoding="utf-8"))
        return JSONResponse(status_code=200, content=data)

    except json.JSONDecodeError as e:
        return JSONResponse(
            status_code=500,
            content={
                "error": "mock.json is not valid JSON",
                "detail": f"{e.msg} (line {e.lineno}, col {e.colno})",
            },
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={"error": "unexpected error reading json file", "detail": str(e)},
        )


# =============================================================================
# POST /postjson
# -----------------------------------------------------------------------------
# 目的:
#   - fetch() POST のデバッグ用。
#   - 「JSONとして正しいか」だけを確認し、内容（フィールド/型/必須）などの検証はしない。
#   - フォーム送信やAPI呼び出しの基本動作確認に使う。
#
# 入力:
#   - Body: JSON（推奨）
#   - Header: Content-Type: application/json（推奨だが、厳密に強制はしない）
#
# 正常:
#   - JSONとしてパース可能 → 200 OK + {"ok": true}
#
# 異常:
#   - 空Body → 400 Bad Request
#   - JSONとして不正（壊れている / JSONではない）→ 400 Bad Request
#
# フロント側の確認例:
#   fetch("/postjson", {method:"POST", headers:{ "Content-Type":"application/json"}, body: JSON.stringify({a:1})})
# =============================================================================
@app.post("/postjson")
async def post_json(request: Request) -> JSONResponse:
    try:
        raw = await request.body()
        if not raw:
            return JSONResponse(status_code=400, content={"error": "empty body (expected JSON)"})

        # JSONとして読めるかのみ確認（構造の意味解釈はしない）
        _parsed: Any = await request.json()

        return JSONResponse(status_code=200, content={"ok": True})

    except Exception:
        return JSONResponse(status_code=400, content={"ok": False, "error": "invalid JSON"})


# =============================================================================
# GET /delay
# -----------------------------------------------------------------------------
# 目的:
#   - 通信遅延をシミュレートし、ローディング表示やタイムアウト設計を確認する。
#   - 「待機中のUI」「キャンセル」「二重送信防止」「スピナー」等のデバッグに使う。
#
# 入力:
#   - なし（固定で2秒遅延）
#
# 出力:
#   - 2秒待機後に 200 OK
#   - {"message":"delayed response","delaySeconds":2}
#
# フロント側の確認例:
#   setLoading(true); await fetch("/delay"); setLoading(false);
# =============================================================================
@app.get("/delay")
async def delay_response() -> JSONResponse:
    await asyncio.sleep(2)
    return JSONResponse(status_code=200, content={"message": "delayed response", "delaySeconds": 2})


# =============================================================================
# GET /items
# -----------------------------------------------------------------------------
# 目的:
#   - ページネーションUIの動作確認（一覧 + ページ送り）。
#   - page/size の扱い、総件数(total)、次ページ有無判定などをフロント側で確認する。
#
# クエリパラメータ:
#   - page: 1以上（例: /items?page=1）
#   - size: 1〜50（例: /items?size=10）
#
# データ仕様:
#   - total = 100 固定の疑似データを返す
#   - items は {"id": number, "name": string} の配列
#
# 出力例:
#   {
#     "page": 2,
#     "size": 10,
#     "total": 100,
#     "items": [ {id:11,name:"Item 11"}, ... {id:20,name:"Item 20"} ]
#   }
#
# 注意:
#   - pageが範囲外でも空配列を返す（UI側の境界処理テストに便利）
# =============================================================================
@app.get("/items")
def get_items(
    page: int = Query(1, ge=1, description="1-based page number"),
    size: int = Query(10, ge=1, le=50, description="page size (1..50)"),
) -> JSONResponse:
    total_items = 100
    all_items: List[Dict[str, Any]] = [{"id": i, "name": f"Item {i}"} for i in range(1, total_items + 1)]

    start = (page - 1) * size
    end = start + size
    paged_items = all_items[start:end]

    return JSONResponse(
        status_code=200,
        content={"page": page, "size": size, "total": total_items, "items": paged_items},
    )


# =============================================================================
# GET /unauthorized
# -----------------------------------------------------------------------------
# 目的:
#   - 401 Unauthorized のUI確認用。
#   - 「未ログイン」扱い時の挙動（ログイン画面へ誘導、トークン再取得、メッセージ表示等）を確認する。
#
# 典型的なフロント側処理:
#   - status === 401 → login画面へリダイレクト / refresh token 実行 / 再ログイン要求
#
# 出力:
#   - 401 Unauthorized
#   - {"error":"Unauthorized"}
# =============================================================================
@app.get("/unauthorized")
def unauthorized() -> JSONResponse:
    return JSONResponse(status_code=401, content={"error": "Unauthorized"})


# =============================================================================
# GET /forbidden
# -----------------------------------------------------------------------------
# 目的:
#   - 403 Forbidden のUI確認用。
#   - 「ログイン済みだが権限がない」ケースの挙動（権限不足表示、申請導線等）を確認する。
#
# 典型的なフロント側処理:
#   - status === 403 → 権限不足ページ/ダイアログ表示
#
# 出力:
#   - 403 Forbidden
#   - {"error":"Forbidden"}
# =============================================================================
@app.get("/forbidden")
def forbidden() -> JSONResponse:
    return JSONResponse(status_code=403, content={"error": "Forbidden"})


# =============================================================================
# GET /notfound
# -----------------------------------------------------------------------------
# 目的:
#   - 404 Not Found のUI確認用。
#   - API側で「存在しないリソース」を返した時の扱い（404ページ、空状態、再試行UI）を確認する。
#
# 注意:
#   - FastAPIの"存在しないパス"にアクセスした場合も 404 になるが、
#     それは「ルーティング不一致の404」。
#   - この /notfound は「APIは存在するが、意図的に404を返す」ため、
#     フロントのハンドリングを確実に再現できる。
#
# 出力:
#   - 404 Not Found
#   - {"error":"Resource not found"}
# =============================================================================
@app.get("/notfound")
def not_found() -> JSONResponse:
    return JSONResponse(status_code=404, content={"error": "Resource not found"})

# =============================================================================
# POST /echo
# -----------------------------------------------------------------------------
# 目的:
#   - fetch() の POST デバッグをさらに楽にする「エコーAPI」。
#   - フロントから送ったJSONが、
#       1) そのままサーバに届いているか
#       2) Content-Type や body の作り方が正しいか
#       3) 文字コードや改行、null/boolean/number の扱いが崩れていないか
#     を最短で確認できる。
#
# 典型的な使い方（フロント側）:
#   - フォーム入力値を JSON.stringify() して送る
#   - 返ってきたJSONを画面に表示し、送信内容と一致しているか確認する
#
# 入力:
#   - Method: POST
#   - Body: JSON（必須）
#   - Header: Content-Type: application/json 推奨
#     ※ただし現場で Content-Type が揺れることもあるため、
#       本APIは「JSONとして解釈できるか」を優先して判定する。
#
# 正常時:
#   - JSONとしてパース可能な本文が来た場合
#   - 200 OK
#   - Body:
#       {
#         "ok": true,
#         "received": (送ったJSONをそのまま)
#       }
#
# 異常時:
#   - 空Body
#       400 Bad Request
#       {"ok": false, "error": "empty body (expected JSON)"}
#   - JSONとして不正（壊れている / JSONではない）
#       400 Bad Request
#       {"ok": false, "error": "invalid JSON"}
#
# 重要:
#   - このAPIは「JSONの中身の検証（必須項目/型/範囲など）」はしない。
#     あくまで fetch の疎通と、送信したJSONが正しく届いているかの確認用途。
# =============================================================================
@app.post("/echo")
async def echo_json(request: Request) -> JSONResponse:
    try:
        raw = await request.body()
        if not raw:
            return JSONResponse(
                status_code=400,
                content={"ok": False, "error": "empty body (expected JSON)"},
            )

        # JSONとしてパースできるかのみ確認（内容の意味解釈はしない）
        data: Any = await request.json()

        # 送られてきたJSONをそのまま返す（デバッグ用途）
        return JSONResponse(
            status_code=200,
            content={"ok": True, "received": data},
        )

    except Exception:
        return JSONResponse(
            status_code=400,
            content={"ok": False, "error": "invalid JSON"},
        )

[
  { "id": 1,  "name": "Item 1",  "category": "A", "price": 1000, "inStock": true  },
  { "id": 2,  "name": "Item 2",  "category": "B", "price": 1100, "inStock": false },
  { "id": 3,  "name": "Item 3",  "category": "C", "price": 1200, "inStock": true  },
  { "id": 4,  "name": "Item 4",  "category": "A", "price": 1300, "inStock": true  },
  { "id": 5,  "name": "Item 5",  "category": "B", "price": 1400, "inStock": false },
  { "id": 6,  "name": "Item 6",  "category": "C", "price": 1500, "inStock": true  },
  { "id": 7,  "name": "Item 7",  "category": "A", "price": 1600, "inStock": true  },
  { "id": 8,  "name": "Item 8",  "category": "B", "price": 1700, "inStock": false },
  { "id": 9,  "name": "Item 9",  "category": "C", "price": 1800, "inStock": true  },
  { "id": 10, "name": "Item 10", "category": "A", "price": 1900, "inStock": true  },
  { "id": 11, "name": "Item 11", "category": "B", "price": 2000, "inStock": false },
  { "id": 12, "name": "Item 12", "category": "C", "price": 2100, "inStock": true  },
  { "id": 13, "name": "Item 13", "category": "A", "price": 2200, "inStock": true  },
  { "id": 14, "name": "Item 14", "category": "B", "price": 2300, "inStock": false },
  { "id": 15, "name": "Item 15", "category": "C", "price": 2400, "inStock": true  },
  { "id": 16, "name": "Item 16", "category": "A", "price": 2500, "inStock": true  },
  { "id": 17, "name": "Item 17", "category": "B", "price": 2600, "inStock": false },
  { "id": 18, "name": "Item 18", "category": "C", "price": 2700, "inStock": true  },
  { "id": 19, "name": "Item 19", "category": "A", "price": 2800, "inStock": true  },
  { "id": 20, "name": "Item 20", "category": "B", "price": 2900, "inStock": false }
]

4) テストサーバー起動（開発用）

uvicorn main:app --reload --host 127.0.0.1 --port 8000

• --reload：ファイル変更時に自動再起動（開発向け）
• ポート競合するなら --port 8001 などに変更

（起動とブラウザ確認の流れはこの手順と同じです）:contentReference[oaicite:2]{index=2}

5) ブラウザを用いた確認

• Hello確認：http://127.0.0.1:8000/hello
• Swagger UI：http://127.0.0.1:8000/docs
• ReDoc：http://127.0.0.1:8000/redoc

（VS Code 補足）Interpreter が .venv になっているか

VS Code 右下（または Ctrl+Shift+P → “Python: Select Interpreter”）で .venv を選んでおくと、実行・補完・デバッグが安定します。:contentReference[oaicite:3]{index=3}