Web API の作成
〜 FastAPI と Uvicorn で作る API 〜
2026-06-26 作成 福島
TOP > tips > fastapi
[ TIPS | TOYS | OTAKU | LINK | MOVIE | CGI | AvTitle | ConfuTerm | HIST | AnSt | Asob | Shell | GBC | LLM ]

0. 前置き

21 世紀を 1/4 を経過する今では、Web API が標準的な業務インタフェースとなりました。
本稿は Python の FastAPI と Uvicorn を使用してじゃんけんの Web API を実装します。

FastAPI は ASGI の関数群を実装するためのヘルパーモジュールであり、単体で動作することができません。
Web API として動作するために HTTP 変換器である Uvicorn (デファクトの ASGI サーバ) を使います。

Linux は実務サーバを考慮して Alma/Rocky を使います。
(複数の Linux ディストリビューションをインストールしても、WSL ではポート共有の動作になることに注意)

動作環境
項目内容備考
WindowsWindows11 pro 25H2本稿記述時の最新版
WSLWSL2
LinuxAlmaLinux release 9.8 (Olive Jaguar)最新の 10 は WSL2 でサポートしていない
言語Python 3.9.25本稿記述時の最新版
API 実装FastAPI 0.128.8
ASGI サーバUvirorn 0.39.0
データベースSQLite3 2.6.0

1. AlmaLinux のインストール

1-1. WSL2 を有効にする。
すでに WSL2 が有効なら実施しない。→ 1-2 へ。

− □ × 
 >_ Windows PowerShell ×   + | ∨
 
Windows PowerShell
Copyright (C) Microsoft Corporation. All rights reserved.

新機能と改善のために最新の PowerShell をインストールしてください!https://aka.ms/PSWindows  

PS C:\> # WSL2 を有効にする
PS C:\> wsl --install --no-distribution ⏎
ダウンロード中: Linux 用 Windows サブシステム 2.7.8
[==========================80.0%=============              ]
インストール中: Linux 用 Windows サブシステム 2.7.8
Linux 用 Windows サブシステム 2.7.8 はインストールされました。
Windows オプション コンポーネントをインストールしています: VirtualMachinePlatform

展開イメージのサービスと管理ツール
バージョン: 10.0.26100.5074

イメージのバージョン: 10.0.26100.8328

機能を有効にしています
[==========================100.0%==========================]
操作は正常に完了しました。
要求された操作は正常に終了しました。変更を有効にするには、システムを再起動する必要があります。
要求された操作は正常に終了しました。変更を有効にするには、システムを再起動する必要があります。

PS C:\> # PC を再起動させる
PS C:\> shutdown /r /t 0 ⏎


ここで PC が再起動する
1-2. WSL2 に Linux をインストールする。
AlmaLinux-9 をインストールする。

− □ × 
 >_ Windows PowerShell ×   + | ∨
 
Windows PowerShell
Copyright (C) Microsoft Corporation. All rights reserved.

新機能と改善のために最新の PowerShell をインストールしてください!https://aka.ms/PSWindows  

PS C:\> # AlmaLinux-9 をインストールする
PS C:\> wsl --install AlmaLinux-9 ⏎
ダウンロードしています: AlmaLinux OS 9
[==========================80.0%=============              ]
ダウンロードしています: AlmaLinux OS 9
インストールしています: AlmaLinux OS 9
[==========================100.0%==========================]
ディストリビューションが正常にインストールされました。'wsl.exe -d AlmaLinux-9' を使用して起動
できます
AlmaLinux-9 を起動しています...
Please create a default UNIX user account. The username does not need to match your Windows u
sername.
For more information visit: https://aka.ms/wslusers
Enter new UNIX username: who
Changing password for user who.
New password: password         ← 実際は表示されない
Retype new password: password
passwd: all authentication tokens updated successfully.


AlmaLinux-9 へ自動的にログインされる

[who@pc who]$ # Linux バージョンを確認する
[who@pc who]$ cat /etc/system-release
AlmaLinux release 9.8 (Olive Jaguar)

[who@pc who]$ # ホスト名を変更する (1)
[who@pc who]$ sudo bash -c "cat << EOF >> /etc/wsl.conf

[network]
hostname = pc
generateHosts = false
EOF"

[who@pc who]$ # ホスト名を変更する (2)
[who@pc who]$ sudo bash -c "echo '127.0.0.1 pc' >> /etc/hosts"

[who@pc who]$ # ssh サーバをインストールする
[who@pc who]$ sudo dnf install -y openssh-server
[who@pc who]$ sudo systemctl start sshd
[who@pc who]$ systemctl status sshd
● sshd.service - OpenSSH server daemon
     Loaded: loaded (/usr/lib/systemd/system/sshd.service; enabled; preset: enanabled)
     Active: active (running) since Sat 2026-06-20 17:39:55 JST; 2h 5min ago
       Docs: man:sshd(8)
             man:sshd_config(5)
   Main PID: 42 (sshd)
      Tasks: 1 (limit: 49304)
     Memory: 5.6M (peak: 7.9M)
        CPU: 261ms
     CGroup: /system.slice/sshd.service
             mq42 "sshd: /usr/sbin/sshd -D [listener] 0 of 10-100 startups"

Jun 20 17:39:55 pc systemd[1]: Starting OpenSSH server daemon...
Jun 20 17:39:55 pc sshd[42]: Server listening on 0.0.0.0 port 22.
Jun 20 17:39:55 pc sshd[42]: Server listening on :: port 22.
Jun 20 17:39:55 pc systemd[1]: Started OpenSSH server daemon.
Jun 20 17:40:16 pc sshd-session[368]: Accepted password for who from 172.31.48.1 port 56973 >
Jun 20 17:40:16 pc sshd-session[368]: pam_unix(sshd:session): session opened for user who(ui>

[who@pc who]$ # AlmaLinux-9 からログアウトする
[who@pc who]$ exit

PS C:\> # AlmaLinux-9 を終了する (→ 次回起動時にホスト名が有効になる)
PS C:\> wsl --shutdown ⏎
PS C:\> exit ⏎

2. Teraterm から AlmaLinux-9 を操作

PoorShell での WSL 操作は非常に使いにくい。
このため、WSL に Teraterm でアクセスする。
しかし WSL は、プロセスが何もないと自動停止 (スリープ) する。

自動停止を回避するために WSL で Linux のダミープロセスを実行してからターミナルソフト (ふつうは Teraterm) から操作する。
作業が終了したらダミープロセスを中断することにより WSL を停止する。

− □ × 
 >_ Windows PowerShell ×   + | ∨
 
Windows PowerShell
Copyright (C) Microsoft Corporation. All rights reserved.

新機能と改善のために最新の PowerShell をインストールしてください!https://aka.ms/PSWindows  

PS C:\> # AlmaLinux-9 の IP アドレスを確認する
PS C:\> wsl -- hostname -I ⏎
172.31.55.38    ← このアドレスに対してすべてのアクセスを行う*1

PS C:\> # 終了しないプロセスで WSL2 を実行する
PS C:\> wsl -- sleep infinity ⏎




ここで Teraterm で 172.31.55.38 に接続し Linux を操作する*2
→ 下記 3 へ



 によってダミープロセスを終了させる (→ WSL2 の停止には数秒かかる)

PS C:\> wsl -l -v ⏎
  NAME           STATE           VERSION
* AlmaLinux-9    Stopped         2

PS C:\> exit ⏎
*1WSL2 では作成された Linux ごとに IP アドレスが自動的に割り振られる。指定はできないが、再起動しても変化しない。
*2具体的には以下を実行する。(パスワードの指定はできない)
 PS C:\> & 'C:\Program Files (x86)\teraterm\ttermpro.exe' who@172.31.55.38 /ssh

3. Python 実行環境を用意
Python と pip を用意する。
[who@pc ~]$ sudo dnf install -y python3 python3-pip

モジュール FastAPI と Uvicorn (HTTP ⇔ ASGI 変換器) を用意する。
[who@pc ~]$ pip install fastapi uvicorn

それぞれのバージョンを確認
[who@pc ~]$ python --version ; pip --version
Python 3.9.25
pip 21.3.1 from /usr/lib/python3.9/site-packages/pip (python 3.9)
[who@pc ~]$ pip show fastapi uvicorn | grep Version
Version: 0.128.8
Version: 0.39.0
[who@pc ~]$ python -c 'import sqlite3 ; print(sqlite3.version)'
2.6.0

4. Web API を作成

以下の仕様で各 Web API を作成する。
API-1:/api/loginログイン
ログインしてセッション ID (UUID) を発行する。
API-2:*3/api/logoutログアウト
ログアウトしてセッション ID を無効化する。
API-3:*3/api/jankenじゃんけん
じゃんけんして勝敗結果をデータベースに残す。
API-4:*3/api/summary戦績取得
データベースから勝敗結果を取り出す。
*3 API-2 ～ API-4 は、API-1 で発行したセッション ID をもとにして動作する。

プログラムは以下。

< jankenapi.py >
import uuid
import random
import sqlite3
from typing import Optional
from fastapi import FastAPI, HTTPException, Header, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from contextlib import asynccontextmanager

API_TITLE = 'Janken API'
EXPIRE_SEC = 60 * 5



def create_db(db_file: str):
    """ データベースを作成する
    """
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    # ユーザーマスタを作成
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            userid TEXT PRIMARY KEY,
            password TEXT NOT NULL
        )
    """)

    # セッション管理テーブルを作成
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS sessions (
            session_id TEXT PRIMARY KEY,
            username   TEXT NOT NULL,
            last_activity TEXT DEFAULT (DATETIME('now', 'localtime'))
        )
    """)

    # じゃんけん戦績テーブルを作成
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS game_records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username  TEXT NOT NULL,
            user_hand TEXT NOT NULL,
            cpu_hand  TEXT NOT NULL,
            result    TEXT NOT NULL,
            played_at TEXT DEFAULT (DATETIME('now', 'localtime'))
        )
    """)

    conn.commit()
    conn.close()



def add_users(db_file: str, users: tuple[tuple[str,str], ...]):
    """ ユーザ登録をする
    """
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    cursor.execute('SELECT COUNT(*) FROM users')
    if cursor.fetchone()[0] == 0:
        cursor.executemany(
                'INSERT INTO users (userid, password) VALUES (?, ?)', users)

    conn.commit()
    conn.close()



def get_username_by_session(db_file: str, session_id: Optional[str]) -> str:
    """ セッション ID の取得・延長をする。
    """
    if not session_id:
        raise HTTPException(status_code=401,
                detail='Session ID is required in Header (Session-Id)')

    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute("""
        SELECT username,
            (strftime('%s', 'now', 'localtime') - strftime('%s', last_activity))
            as passed_sec
        FROM sessions WHERE session_id = ?
    """, (session_id,))

    row = cursor.fetchone()

    if not row:
        conn.close()
        raise HTTPException(status_code=401,
                detail='Invalid or expired session')

    username, passed_sec = row

    # 5分以上経過していたらセッションを無効化（削除）して弾く
    if passed_sec >= EXPIRE_SEC:
        cursor.execute('DELETE FROM sessions WHERE session_id = ?', (session_id,))
        conn.commit()
        conn.close()
        raise HTTPException(status_code=401,
                detail='Session expired due to inactivity')

    # 5分以内なら、最終アクティブ時間を更新 (セッション寿命を延長)
    cursor.execute("""
        UPDATE sessions
        SET last_activity = DATETIME('now', 'localtime')
        WHERE session_id = ?
    """, (session_id,))

    conn.commit()
    conn.close()

    return username



@asynccontextmanager
async def lifespan(app: FastAPI):
    """起動時/終了時の処理
    """
    print(f'{API_TITLE} を起動します。')

    app.state.db_file = './system.db'
    create_db(app.state.db_file)
    add_users(app.state.db_file, (('user1','password1'),))

    yield

    print(f'{API_TITLE} を終了します。')

# FastAPI の初期化
app = FastAPI(
    title = 'Janken API',   # /docs に表示するタイトル
    lifespan = lifespan,    # 起動時/終了時に実行する関数
    root_path = '/api'      # /api/* でアクセスされる前提*4
)

# CORS 設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)



class LoginRequest(BaseModel):
    userid: str
    password: str

class JankenRequest(BaseModel):
    hand: str



@app.post('/login')
def login(req_body: LoginRequest, request: Request):
    """ API-1. ログイン
    """
    db_file = request.app.state.db_file
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    try:
        # 5 分以上操作されていないセッショントークンを削除
        cursor.execute(
        f"""DELETE FROM sessions
        WHERE last_activity < DATETIME('now', 'localtime', '-{EXPIRE_SEC} seconds')
        """)
    except sqlite3.IntegrityError:
        pass
    finally:
        conn.commit()

    # ユーザ ID からパスワードを取得
    cursor.execute('SELECT password FROM users WHERE userid = ?',
            (req_body.userid,))
    row = cursor.fetchone()

    # パスワードを照合
    if not row or row[0] != req_body.password:
        conn.close()
        raise HTTPException(status_code=401, detail='Incorrect userid or password')

    try:
        # セッショントークンを発行
        session_id = str(uuid.uuid4())
        session_id = 'dummyfor-test-0000-1111-abcdefghijkl' # テスト用ダミー*5
        cursor.execute(
            'INSERT INTO sessions (username, session_id) VALUES (?, ?)',
            (req_body.userid, session_id)
        )
    except sqlite3.IntegrityError:
        return {
            'status': 'error',
            'description': 'User denied'
        }
    finally:
        conn.commit()
        conn.close()

    return {
        'session_id': session_id,
        'username':   req_body.userid,
    }



@app.post('/logout')
def logout(request: Request, session_id: Optional[str] = Header(None)):
    """ API-2. ログアウト
    """
    db_file = request.app.state.db_file
    get_username_by_session(db_file, session_id)

    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    cursor.execute('DELETE FROM sessions WHERE session_id = ?',
            (session_id,))
    conn.commit()
    conn.close()

    return {'message': 'Logged out successfully'}



@app.post('/janken')
def janken(req_body: JankenRequest, request: Request,
        session_id: Optional[str] = Header(None)
        ):
    """ API-3. じゃんけん
    """
    db_file = request.app.state.db_file
    username = get_username_by_session(db_file, session_id)

    user_hand = req_body.hand
    valid_hands = [
            'Rock',     # グー
            'Scissors', # チョキ
            'Paper'     # パー
    ]

    if user_hand not in valid_hands:
        raise HTTPException(status_code=400,
                detail="Hand must be 'Rock', 'Scissors', or 'Paper'")

    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    # 今回のじゃんけんを実行
    cpu_hand = random.choice(valid_hands)
    if user_hand == cpu_hand:
        result = 'Draw'
    elif (user_hand == 'Rock'     and cpu_hand == 'Scissors') \
      or (user_hand == 'Scissors' and cpu_hand == 'Paper') \
      or (user_hand == 'Paper'    and cpu_hand == 'Rock'):
        result = 'Win'
    else:
        result = 'Lose'

    # 勝敗結果を保存
    cursor.execute("""
        INSERT INTO game_records (username, user_hand, cpu_hand, result)
        VALUES (?, ?, ?, ?)
    """, (username, user_hand, cpu_hand, result))

    conn.commit()
    conn.close()

    return {
        'user_hand': user_hand,
        'cpu_hand':  cpu_hand,
        'result':    result,
    }



@app.post('/summary')
def get_summary(request: Request,
        session_id: Optional[str] = Header(None)):
    """ API-4. 前回までの戦績を取得
    """
    db_file = request.app.state.db_file
    username = get_username_by_session(db_file, session_id)

    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()

    # 勝ち/負け/引き分け、それぞれの総数を求める
    cursor.execute("""
        SELECT
            COUNT(CASE WHEN result = 'Win'  THEN 1 END) as win,
            COUNT(CASE WHEN result = 'Lose' THEN 1 END) as lose,
            COUNT(CASE WHEN result = 'Draw' THEN 1 END) as draw
        FROM game_records
        WHERE username = ?
    """, (username,))

    win, lose, draw = cursor.fetchone()
    conn.close()

    return {'win': win, 'lose': lose, 'draw': draw}
*4 Proxy から呼ばれるときの補正。API は /* でも動作する。
*5テストを簡易にするために入れている。簡易テストが終了したら、この行は取り除いてテストすること。

5. 動作確認

5-1. Web API の起動
jankenapi.py を実行する。

$ uvicorn jankenapi:app --host 0.0.0.0 --port 8000
5-2. ログイン / ログアウトを確認。
5-2-1. ログインする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/login \
  -H "Content-Type: application/json" \
  -d '{"userid":"user1", "password":"password1"}' ; echo
PoorShell の場合は '{\"userid\":\"user1\", \"password\":\"password1\"}' のように、ダブルクォートをすべてエスケープする。

• レスポンス
{"session_id":"dummyfor-test-0000-1111-abcdefghijkl","username":"user1"}
5-2-2. ログアウトする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/logout \
  -H "Session-Id: dummyfor-test-0000-1111-abcdefghijkl" ; echo

• レスポンス
{"message":"Logged out successfully"}
5-3. じゃんけんを実施
5-3-1. ログインする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/login \
  -H "Content-Type: application/json" \
  -d '{"userid":"user1", "password":"password1"}' ; echo

• レスポンス
{"session_id":"dummyfor-test-0000-1111-abcdefghijkl","username":"user1"}
5-3-2. じゃんけんを実行する。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/janken \
  -H "Content-Type: application/json" \
  -H "Session-Id: dummyfor-test-0000-1111-abcdefghijkl" \
  -d '{"hand":"Rock"}' ; echo

• レスポンス
{"user_hand":"Rock","cpu_hand":"Scissors","result":"Win"}
5-3-3. ログアウトする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/logout \
  -H "Session-Id: dummyfor-test-0000-1111-abcdefghijkl" ; echo

• レスポンス
{"message":"Logged out successfully"}
5-4. 戦績を確認
5-4-1. ログインする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/login \
  -H "Content-Type: application/json" \
  -d '{"userid":"user1", "password":"password1"}' ; echo

• レスポンス
{"session_id":"dummyfor-test-0000-1111-abcdefghijkl","username":"user1"}
5-4-2. 戦績を確認する。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/summary \
  -H "Session-Id: dummyfor-test-0000-1111-abcdefghijkl" ; echo

• レスポンス
{"win":1,"lose":0,"draw":0}
5-4-3. ログアウトする。
• リクエスト
$ curl -X POST 172.31.55.38:8000/api/logout \
  -H "Session-Id: dummyfor-test-0000-1111-abcdefghijkl" ; echo

• レスポンス
{"message":"Logged out successfully"}