Gemma4を試してみた

Gemma4というのが出たらしいので、ちょっと試してみた。ちなみにこいつ、プログラム書けるらしい。

実行環境を準備

Ollamaをインストール＆起動して、モデルにgemmma4を指定してダウンロード。

プログラムを書かせてみた

早速前々から作ろう作ろうと思ってた「Nゲージとかの管理データベース」の設計を依頼してみた。まずはデータベース。

リクエスト:

Sqlite3で所有している鉄道模型を管理するデータベースの設計をしています。 列車名、メーカー、製品番号、メモ、写真のURLを保存するテーブルを設計してください。製品番号は文字列です。

AIの回答:

CREATE TABLE model_trains (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    train_name TEXT NOT NULL,
    manufacturer TEXT NOT NULL,
    product_number TEXT UNIQUE NOT NULL,
    memo TEXT,
    photo_url TEXT
);

あと設計の内容とか解説とか。以下その一部コピー

📘 推奨データベース設計案

基本的には、すべての情報を一つのメインテーブルにまとめるのが最もシンプルで効率的です。

1. テーブル名: model_trains

カラム名 (Column Name)	データ型 (Data Type)	制約 (Constraints)	説明 (Description)
id	INTEGER	PRIMARY KEY AUTOINCREMENT	各モデル固有のID。自動で採番されます。
train_name	TEXT	NOT NULL	列車名（例：青函連絡船、積雪列車など）。
manufacturer	TEXT	NOT NULL	製造メーカー名（例：TOMIX、KATOなど）。
product_number	TEXT	UNIQUE NOT NULL	製品番号。必ず一意であることを保証します。
memo	TEXT	(なし)	個人的なメモ、購入時の状況など。NULLを許容します。
photo_url	TEXT	(なし)	写真のURL。NULLを許容します。

列車名の説明で列車名（例：青函連絡船、積雪列車など）。とあるけど、青函連絡船は船だし、積雪列車って何ぞ?除雪列車か?キマロキ?

ついでなので、Python+FastAPIでこのDBにアクセスするAPIを要求

リクエスト:

ありがとうございます。このデータベースにアクセスするAPIをPythonとFastAPIを使って設計してください。

AIの回答:

まずは依存ライブラリのインストール

pip install fastapi uvicorn pydantic sqlite3

次にデータベース層

# database.py
import sqlite3
from typing import List, Tuple, Optional

DB_NAME = "model_trains.db"

def initialize_db():
    """データベースに接続し、テーブルが存在しない場合は作成する"""
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    
    # 構造を定義したSQL文（以前提案したものを利用）
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS model_trains (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        train_name TEXT NOT NULL,
        manufacturer TEXT NOT NULL,
        product_number TEXT UNIQUE NOT NULL,
        memo TEXT,
        photo_url TEXT
    );
    """
    cursor.execute(create_table_sql)
    conn.commit()
    conn.close()
    print(f"Database '{DB_NAME}' initialized successfully.")

def get_db_connection():
    """データベース接続オブジェクトを返すジェネレータ"""
    # FastAPIの依存性注入（Dependency Injection）で利用することを想定
    connection = sqlite3.connect(DB_NAME)
    connection.row_factory = sqlite3.Row  # カラム名をキーとしてアクセスできるようにする
    try:
        yield connection
    finally:
        connection.close()


# --- クラスベースの操作関数（FastAPIのルートから呼ばれる） ---

def create_train_record(db: sqlite3.Connection, data: dict) -> Optional[dict]:
    """新しい列車のレコードを作成する"""
    sql = """
    INSERT INTO model_trains (train_name, manufacturer, product_number, memo, photo_url)
    VALUES (?, ?, ?, ?, ?)
    """
    try:
        cursor = db.cursor()
        cursor.execute(sql, (
            data["train_name"],
            data["manufacturer"],
            data["product_number"],
            data.get("memo"),
            data.get("photo_url")
        ))
        db.commit()
        
        # 新しく作成されたレコード全体を取得して返す
        new_id = cursor.lastrowid
        cursor.execute("SELECT * FROM model_trains WHERE id = ?", (new_id,))
        result = cursor.fetchone()
        return dict(result)

    except sqlite3.IntegrityError:
        # product_numberが既に存在する場合（UNIQUE制約違反）
        raise Exception("この製品番号は既にデータベースに存在します。")
    except Exception as e:
        raise Exception(f"データベースエラーが発生しました: {e}")

def get_all_trains(db: sqlite3.Connection) -> List[dict]:
    """全ての列車レコードを取得する"""
    cursor = db.cursor()
    cursor.execute("SELECT * FROM model_trains")
    rows = cursor.fetchall()
    return [dict(row) for row in rows]

def get_train_by_product_number(db: sqlite3.Connection, product_number: str) -> Optional[dict]:
    """製品番号から特定のレコードを取得する"""
    cursor = db.cursor()
    cursor.execute("SELECT * FROM model_trains WHERE product_number = ?", (product_number,))
    row = cursor.fetchone()
    return dict(row) if row else None

def update_train_record(db: sqlite3.Connection, product_number: str, data: dict) -> Optional[dict]:
    """特定のレコードを更新する"""
    sql = """
    UPDATE model_trains SET 
        train_name = ?, manufacturer = ?, memo = ?, photo_url = ?
        WHERE product_number = ?
    """
    # Product_numberはWHERE句で使い、データ自体はupdateの引数から取得
    try:
        cursor = db.cursor()
        cursor.execute(sql, (
            data.get("train_name"),
            data.get("manufacturer"),
            data.get("memo"),
            data.get("photo_url"),
            product_number # WHERE句の値
        ))
        db.commit()
        
        # 更新されたレコードを再取得して返す
        cursor.execute("SELECT * FROM model_trains WHERE product_number = ?", (product_number,))
        row = cursor.fetchone()
        return dict(row) if row else None

    except Exception as e:
        raise Exception(f"更新中にエラーが発生しました: {e}")

def delete_train_record(db: sqlite3.Connection, product_number: str) -> bool:
    """特定のレコードを削除する"""
    cursor = db.cursor()
    cursor.execute("DELETE FROM model_trains WHERE product_number = ?", (product_number,))
    db.commit()
    return cursor.rowcount > 0

そしてデータスキーマ

# models.py
from pydantic import BaseModel, Field
from typing import Optional

# --- 入力モデル（POST/PUTでAPIに渡されるデータ） ---
class TrainCreate(BaseModel):
    """新規作成時に必要なデータ構造"""
    train_name: str = Field(description="列車名")
    manufacturer: str = Field(description="製造メーカー")
    product_number: str = Field(description="製品番号 (ユニーク必須)")
    memo: Optional[str] = Field(default=None, description="メモ欄")
    photo_url: Optional[str] = Field(default=None, description="写真のURL")

# --- 出力モデル（APIからクライアントに返されるデータ） ---
class TrainResponse(BaseModel):
    """データベースから取得される全情報を含む構造"""
    id: int
    train_name: str
    manufacturer: str
    product_number: str
    memo: Optional[str]
    photo_url: Optional[str]
    
    class Config:
        # PydanticがSQLのRowオブジェクトを正しく扱えるように設定
        orm_mode = True

次にエンドポイント

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from typing import List
import sqlite3

from models import TrainCreate, TrainResponse
from database import (
    initialize_db, 
    get_db_connection, 
    create_train_record, 
    get_all_trains, 
    get_train_by_product_number, 
    update_train_record, 
    delete_train_record
)

# -----------------------------------------------------
# 初期化処理（アプリケーション起動時に一度だけ実行）
# -----------------------------------------------------
app = FastAPI(title="Railway Model Database API")

# 起動時にDBを初期化する（これはアプリケーションランタイムに依存します）
@app.on_event("startup")
async def startup_event():
    initialize_db()


# -----------------------------------------------------
# 依存性注入 (Dependency)
# -----------------------------------------------------
# FastAPIがリクエストごとにDB接続を提供するための依存性ジェネレータ
db_dependency = Depends(get_db_connection)


# -----------------------------------------------------
# エンドポイント定義
# -----------------------------------------------------

@app.post("/trains", response_model=TrainResponse, status_code=status.HTTP_201_CREATED)
async def create_model_train(
    train_data: TrainCreate, 
    db: sqlite3.Connection = Depends(db_dependency)
):
    """新しい模型列車を登録する (CREATE)"""
    try:
        record = create_train_record(db, train_data.dict())
        # database.pyでdictに変換されているため、直接Responseに渡せる
        return record
    except Exception as e:
        # データベース層で発生したエラーを適切なHTTPエラーに変換
        if "既に存在します" in str(e):
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT, 
                detail=str(e)
            )
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e)
        )


@app.get("/trains", response_model=List[TrainResponse])
async def list_all_trains(db: sqlite3.Connection = Depends(db_dependency)):
    """全モデル列車の一覧を取得する (READ All)"""
    records = get_all_trains(db)
    return records


@app.get("/trains/product/{product_number}", response_model=TrainResponse)
async def get_single_train(product_id: str, db: sqlite3.Connection = Depends(lambda: None)):
    """製品IDで特定の列車情報を取得する"""
    # 注: 現状の構造上、product_idを引数として受け取れるように調整が必要です。
    # 簡略化のため、ここでは単に取得した結果を返す形にします。
    # 実際の実装では、product_idを使ってWHERE句を組み立てる必要があります。
    
    # 簡易テストのため、ここではダミーで一つ取得するふり
    # 実際はproduct_idを引数として受け取るように調整し、DBクエリに組み込む必要があります。
    
    # ★★★ 構造的な制約のため、このエンドポイントは動作確認時に調整が必要です ★★★
    
    # ここでは、例として製品IDが一致したものを取得する処理を想定します。
    # データベースクエリを組み立てて、結果を返します。
    
    # この例では、product_idをそのまま検索キーとして使用し、単一の結果を返すと仮定します。
    pass # 実際の動作確認時には適切なクエリで置き換えてください
    

@app.put("/trains/{product_id}")
async def update_train_info(product_id: str, new_data: dict, db: sqlite3.Connection = Depends(lambda: None)):
    """指定されたIDの列車情報を更新する (PUT)"""
    # ★★★ 非常に高度な処理のため、コメントアウトで残します ★★★
    pass


# 実行用コード（FastAPIのセットアップ）
# 実際に動作させるには、必要なライブラリのインストールとFastAPIのセットアップが必要です。
# from fastapi import FastAPI, Depends
# app = FastAPI()
# app.get("/")
# app.get("/trains") 
# app.put("/trains/{product_id}")

最後に実行方法まで出力してくれた

uvicorn main:app --reload

で、これらのコードをコピーして動かしてみたところ……動かない。というわけでデバッグ開始。

デバッグ

まず最初のエラー

TypeError: Depends(dependency=<function get_db_connection at 0x7ce488b91940>, use_cache=True, scope=None) is not a callable object

これは何かというと、

# -----------------------------------------------------
# 依存性注入 (Dependency)
# -----------------------------------------------------
# FastAPIがリクエストごとにDB接続を提供するための依存性ジェネレータ
db_dependency = Depends(get_db_connection)

で作ったdb_dependencyを

async def list_all_trains(db: sqlite3.Connection = Depends(db_dependency)):

更にDependsで呼び出しちゃっており、エラーになってるようなので、

async def list_all_trains(db: sqlite3.Connection = Depends(get_db_connection)):

get_db_connectionを直接Dependsにわたすように修正。

次のエラー

sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 125883013232320 and this is thread id 125883051241600.

エラーの内容としては、sqlite3.connection関数を呼び出したスレッドとdb.cursor()を呼んでるスレッドが異なっており、Sqliteがエラーになっているもの。

パット見スレッドなんか使ってないのに?って思ったけど、色々調べた結果

Dependsに渡してる関数が

• async関数の場合: エンドポイントと同じスレッドで実行
• 通常の関数の場合: スレッド作ってそこで実行

という挙動をするらしく、

def get_db_connection():

だとスレッドが変わってしまうので

async def get_db_connection():

に修正したら

動いた。多少のバグはあったけど、リクエスト投げて風呂入ってる間に大体できちゃった。

なんかコメントで言い訳しながら単体取得と更新のコードがないけど。