Gabu Quest
🇺🇸 English

SQLer フィールドガイド ★ SQLerのベンチマーク概要

SQLite と JSON を扱う Pydantic モデル。ネストしたフィルタとリレーション補助で数分で動くデータストアを届けます。

役割: メイン開発 @ サイドプロジェクト
期間: 2025年5月 – 2025年9月
技術:
PythonPydanticSQLiteSQLJSON

本当に使える JSON モデル

またカイゼン。20分で何かを見せる必要がある。ORMやマイグレーションは会議時間を簡単に超えてしまう。SQLer は「いま動くノート」。SQLite 上の JSON を使う Pydantic モデル、Python らしく書けるクエリ、そして儀式抜きのリレーション解決。

要点: SQLer は、JSON を格納先にした Pydantic モデル、ネストしたデータへの直感的なフィルタ、参照の解決、楽観的ロック、バルク upsert、JSON 向けのインデックスを提供します。基盤は SQLite。数時間ではなく数分で「動く保管庫」が欲しいときに効きます。

最初の保存: SQLer の基本

空の SQLite データベースを用意して、日本の都道府県を覚えさせ、Python のオブジェクトのように問い合わせます。

from sqler import SQLerDB, SQLerModel
from sqler.query import SQLerField as F

class Prefecture(SQLerModel):
    name: str
    region: str
    population: int
    foods: list[str] | None = None

class City(SQLerModel):
    name: str
    population: int
    prefecture: Prefecture | None = None

# お手軽: メモリ上のデータベース
db = SQLerDB.in_memory()
Prefecture.set_db(db)
City.set_db(db)

# データ作成
kyoto = Prefecture(name="Kyoto", region="Kansai", population=2_585_000, foods=["matcha", "yudofu"]).save()
osaka = Prefecture(name="Osaka", region="Kansai", population=8_839_000, foods=["takoyaki"]).save()

# 直感的なクエリ
big_prefectures = (
    Prefecture.query()
    .filter(F("population") > 1_000_000)
    .order_by("population", desc=True)
    .all()
)
assert [p.name for p in big_prefectures][:2] == ["Osaka", "Kyoto"]

ポイントは3つ:

  1. モデルは Pydantic — 型ヒント、バリデーション、デフォルトがそのまま使える。
  2. 保存は明示的 — .save() は _id を付けて self を返す。いつ永続化するかは自分で決める。
  3. クエリは Python 的 — F("population") > 1_000_000 は JSON に優しい SQL にコンパイルされる。

テスト: これらの例は契約スイートに含まれます（tests/test_readme.py および tests/sync/**）。

Sync か Async か? — 両方

最初に選ぶ必要はありません。SQLer は並行する API を提供します。

import asyncio
from sqler import AsyncSQLerDB, AsyncSQLerModel
from sqler.query import SQLerField as F

class AUser(AsyncSQLerModel):
    name: str
    age: int

async def main():
    db = AsyncSQLerDB.in_memory()
    await db.connect()
    AUser.set_db(db)

    await AUser(name="Ada", age=36).save()
    adults = await AUser.query().filter(F("age") >= 18).order_by("age").all()
    assert any(u.name == "Ada" for u in adults)

    await db.close()

asyncio.run(main())

非同期 API は同期版と同じ感覚で扱えます。まずは同期で試し、必要になったら FastAPI などへ移行すればよいだけ。

配列クエリで泣かない

「オブジェクトの配列」を SQL 儀式なしで絞り込みたい。.any().where(...) で配列要素の条件を素直に書けます。

from sqler import SQLerDB, SQLerModel
from sqler.query import SQLerField as F

class Order(SQLerModel):
    customer: str
    items: list[dict] | None = None

db = SQLerDB.in_memory()
Order.set_db(db)

# データ投入
Order(customer="Customer1", items=[{"sku": "RamenSet", "qty": 3}, {"sku": "Gyoza", "qty": 1}]).save()
Order(customer="Customer2", items=[{"sku": "RamenSet", "qty": 1}]).save()

# ラーメンの qty >= 2 を含む注文
ramen_lovers = (
    Order.query()
    .filter(F(["items"]).any().where((F("sku") == "RamenSet") & (F("qty") >= 2)))
    .all()
)
assert [o.customer for o in ramen_lovers] == ["Customer1"]

包含・所属・除外・インスペクションも素直に書けます。

from sqler import SQLerDB, SQLerModel
from sqler.query import SQLerField as F

class QueryUser(SQLerModel):
    name: str
    age: int
    tags: list[str] | None = None
    tier: int | None = None

class QueryOrder(SQLerModel):
    customer: str
    items: list[dict] | None = None

db = SQLerDB.in_memory()
QueryUser.set_db(db)
QueryOrder.set_db(db)

QueryUser(name="Ada", age=36, tags=["pro", "python"], tier=1).save()
QueryUser(name="Bob", age=20, tags=["hobby"], tier=3).save()

QueryOrder(customer="Ada", items=[{"sku": "ABC", "qty": 3}]).save()
QueryOrder(customer="Bob", items=[{"sku": "XYZ", "qty": 1}]).save()

pros = QueryUser.query().filter(F("tags").contains("pro")).all()
assert [u.name for u in pros] == ["Ada"]

premium = QueryUser.query().filter(F("tier").isin([1, 2])).all()
assert [u.name for u in premium] == ["Ada"]

real_users = QueryUser.query().exclude(F("name").like("test%")).all()
assert {u.name for u in real_users} == {"Ada", "Bob"}

expr = F(["items"]).any().where((F("sku") == "ABC") & (F("qty") >= 2))
assert [o.customer for o in QueryOrder.query().filter(expr).all()] == ["Ada"]

sql, params = QueryUser.query().filter(F("age") >= 18).debug()
assert isinstance(sql, str) and params == [18]

メモ:

  • Model.query() は .sql() / .params() メソッド と .debug() を持ちます。
  • 低レベルの SQLerQuery は .sql / .params プロパティです。
  • explain_query_plan(adapter) で SQLite の実行計画をのぞけます。

儀式なしのリレーション

参照は JSON で { "_table": "addresses", "_id": 1 } のように保存され、ロード時にモデルへ水和（hydrate）されます。

from sqler import SQLerDB, SQLerModel

class Address(SQLerModel):
    city: str
    country: str

class User(SQLerModel):
    name: str
    address: Address | None = None

db = SQLerDB.in_memory()
Address.set_db(db)
User.set_db(db)

home = Address(city="Kyoto", country="JP").save()
user = User(name="Alice", address=home).save()

loaded = User.from_id(user._id)
assert loaded.address.city == "Kyoto"

kyoto_users = User.query().filter(User.ref("address").field("city") == "Kyoto").all()
assert [u.name for u in kyoto_users] == ["Alice"]

用心深い人のための楽観的ロック

複数プロセスが同じレコードを更新する可能性があるなら、SQLerSafeModel と _version で衝突を検出します。

from sqler import SQLerDB, SQLerSafeModel, StaleVersionError

class Account(SQLerSafeModel):
    owner: str
    balance: int

db = SQLerDB.in_memory()
Account.set_db(db)

account = Account(owner="Ada", balance=100).save()
account.balance = 120
account.save()  # バージョンがインクリメントされる

# 競合を擬似再現: JSON 側の _version を手で加算

db.adapter.execute(
    "UPDATE accounts SET data = json_set(data,'$._version', json_extract(data,'$._version') + 1) WHERE _id = ?",
    [account._id],
)
db.adapter.commit()

# いま手元のコピーは古い
account.balance = 130
try:
    account.save()
except StaleVersionError:
    account.refresh()
    account.balance = 130
    account.save()

FastAPI が未導入でも動く 409 変換の最小例:

try:
    from fastapi import HTTPException
except ImportError:  # docs/CI 用のフォールバック
    class HTTPException(Exception):
        def __init__(self, status_code: int, detail: str):
            self.status_code = status_code
            self.detail = detail

from sqler.models import StaleVersionError

try:
    account.save()
except StaleVersionError:
    raise HTTPException(409, "Version conflict; refresh and retry")

外部キーなしでも参照整合性を守る

削除ポリシーで参照の健全性を保ちます:

  • restrict — 参照されている間は削除を拒否
  • set_null — 参照側のフィールドを削除前に NULL へ
  • cascade — 依存側を再帰的に削除
from sqler import SQLerDB, SQLerModel
from sqler.models import ReferentialIntegrityError

class DIUser(SQLerModel):
    name: str

class Post(SQLerModel):
    title: str
    author: dict | None = None

# restrict — 参照がある限り削除不可
restrict_db = SQLerDB.in_memory()
DIUser.set_db(restrict_db)
Post.set_db(restrict_db)
writer = DIUser(name="Writer").save()
Post(title="Post A", author={"_table": "diusers", "_id": writer._id}).save()
try:
    writer.delete_with_policy(on_delete="restrict")
    assert False, "expected ReferentialIntegrityError"
except ReferentialIntegrityError:
    pass

# set_null — 参照を空にしてから削除
set_null_db = SQLerDB.in_memory()
DIUser.set_db(set_null_db)
Post.set_db(set_null_db)
nullable = DIUser(name="Nullable").save()
post = Post(title="Post B", author={"_table": "diusers", "_id": nullable._id}).save()
nullable.delete_with_policy(on_delete="set_null")
assert Post.from_id(post._id).author is None

# cascade — 依存を巻き取って削除
cascade_db = SQLerDB.in_memory()
DIUser.set_db(cascade_db)
Post.set_db(cascade_db)
cascade = DIUser(name="Cascade").save()
Post(title="Post C", author={"_table": "diusers", "_id": cascade._id}).save()
cascade.delete_with_policy(on_delete="cascade")
assert Post.query().count() == 0

潜在的な「浮遊参照」を検出:

class RefUser(SQLerModel):
    name: str

class RefPost(SQLerModel):
    title: str
    author: dict | None = None

vdb = SQLerDB.in_memory()
RefUser.set_db(vdb)
RefPost.set_db(vdb)

u = RefUser(name="Ada").save()
dangling = RefPost(title="Lost", author={"_table": RefUser.__tablename__, "_id": u._id}).save()

vdb.delete_document(RefUser.__tablename__, u._id)
broken = RefPost.validate_references()

# 返り値は sqler.models.BrokenRef の配列
assert broken and broken[0].row_id == dangling._id

バルク操作と SQL の逃げ道

大量投入が必要なときもあれば、生 SQL が欲しいときもある。

バルク upsert は、指定済みの ID を尊重し、結果の ID を入力順に返します。

from sqler import SQLerDB, SQLerModel

class BulkUser(SQLerModel):
    name: str
    age: int | None = None

db = SQLerDB.in_memory()
BulkUser.set_db(db)

rows = [{"name": "Alice"}, {"name": "Bob"}, {"_id": 999, "name": "Carol"}]
ids = db.bulk_upsert(BulkUser.__tablename__, rows)
assert len(ids) == 3 and 999 in ids

生 SQL（戻りはデコード済みの辞書）:

class ReportUser(SQLerModel):
    name: str
    email: str | None = None

rdb = SQLerDB.in_memory()
ReportUser.set_db(rdb)
ReportUser(name="Ada", email="ada@example.com").save()
ReportUser(name="Bob", email="bob@example.com").save()

rows = rdb.execute_sql(
    """
    SELECT u._id, u.data
    FROM reportusers u
    WHERE json_extract(u.data,'$.name') LIKE ?
    """,
    ["A%"],
)

# execute_sql は JSON を展開して結合した dict を返す（data キーは残らない）
# 例: {"_id": 1, "name": "Ada", "email": "ada@example.com"}
assert rows and rows[0]["_id"] == 1 and "data" not in rows[0]

JSON フィールドへのインデックス（性能対策）:

# 単純なフィールド
db.create_index("users", "age")

# 一意制約
db.create_index("users", "email", unique=True)

# WHERE 付きの部分インデックス
db.create_index("users", "age", where="json_extract(data,'$.age') IS NOT NULL")

# ネストしたフィールド
db.create_index("users", "address.city")

これらは SQLite の JSON1 (json_extract(...)) に展開され、実行計画の助けになります。

デプロイの注意

  • 接続: アプリ側で管理。SQLite は WAL を有効化すると同時アクセスに強くなる。
  • 性能: JSON クエリは正規化テーブルより遅い場合がある。ホットパスにはインデックス。
  • スキーマ進化: フィールド追加は容易。削除や型変更はクリンナップが要ることがある。
  • テスト: 契約スイート（tests/test_readme.py など）でドキュメントと挙動のズレを防ぐ。

次にやること

uv add sqler で入り口は完成。モデルを1つ作って .save()。uv run pytest -q で例が実行されるのを確認。クエリに違和感があれば .debug() が生成 SQL とパラメータを見せてくれる。

SQLer は「いますぐ構造化データが必要」で「モデルはまだ動く」場面のための道具。賢さより明快さを優先し、よく通る道を素直にする。小さく作り、どこで曲がるか確かめよう。

English: Read this project in English