0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

PR: Disnana
Disnana公式サイト
About Promotions
@harupython
in

型保障のPydanticとSQLiteを組み合わせたNyanSQLiteを使ってみた

0
Posted at

「ちょっとしたDB」、どうしてますか？

Pythonで個人開発や社内ツールを作るとき、こんな場面に出くわしたことはありませんか？

「あ、このデータ、次回起動時にも残しておきたい……」

そのとき、あなたはどうしますか？

  • SQLAlchemy を使う → セットアップが重い。Alembicのマイグレーション、モデル定義、セッション管理……小さなツールに対して明らかにオーバースペック
  • 生SQL + sqlite3 → シンプルだが cursor.execute("SELECT * FROM ...") を書くたびに型が消えていく。辞書を返すだけのAPIに疲れてくる
  • JSON / pickle → 最初は楽だが、データが増えてくると検索も更新も地獄。型保証なんて夢のまた夢

「そこそこ型安全で、そこそこ速くて、面倒な設定が要らないDBラッパーがほしい」

そう思い続けて何年経ったか……というエンジニアに届けたいのが、今回紹介する NyanSQLite です。

NyanSQLiteとは何者か

NyanSQLiteを一言で言うと、

PydanticモデルをそのままDBスキーマとして使える、型安全・高性能なSQLiteラッパー

です。

もともと NanaSQLite というKVSベースのライブラリが原点にあり、そのKVS設計を丸ごと捨てて高速化に全振りして再設計されたOSSです。
「NyanSQLite」という名前にはそういう経緯があります。

主な特徴をざっくり並べると、

機能 説明
Pydanticベースのスキーマ 型ヒントで自動バリデーション、JSON変換も透過的
Djangoライクなクエリ __gte、__in、__like など直感的なフィルタリング
FTS5全文検索 Searchable[str] アノテーションだけで全文検索が有効に
自動インデックス管理 Indexed[T] アノテーションでB-treeインデックスを自動構築
パフォーマンス最適化 WALモード・バッチ処理・自動チャンキングで高速化
マルチスレッド / asyncio 対応 threading.Lock + NyanSQLiteAIO で並行処理も安全

「ふーん、よくあるORMラッパーでしょ」と思ったあなた、ちょっと待ってください。
このライブラリの真骨頂は、型ヒントがそのままDBの挙動に直結するところにあります。実際に使い始めると、その体験の気持ちよさに気づくはずです。

5分で動かしてみる

まずインストールから。

pip install nyansqlite

Pydantic v2 が必須です。v1系のプロジェクトでは動作しません。

pip install "pydantic>=2.0"

Python 3.9以上が必要です。

インストールはこれだけ。では実際に動かしてみましょう。

① スキーマ定義（ここが核心）

from pydantic import BaseModel
from nyansqlite import NyanSQLite, Indexed, Searchable

class Article(BaseModel):
    id: int                   # id フィールドが自動的に主キーになる
    author: Indexed[str]      # ← これだけでB-treeインデックスが作られる
    title: Searchable[str]    # ← これだけでFTS5全文検索の対象になる
    body: Searchable[str]
    views: int = 0

Indexed[T] と Searchable[str] がこのライブラリの象徴的な設計です。
型ヒントを書くだけでインデックス作成・全文検索設定まで完結します。
別ファイルにインデックス設定を書く必要も、マイグレーションファイルを管理する必要もありません。

② DB初期化・データ挿入

db = NyanSQLite("blog.db")
db.register(Article)    # ← これだけでCREATE TABLEが走る

db.insert(Article(
    id=1,
    author="neko",
    title="SQLiteを使いこなそう",
    body="NyanSQLiteで簡単にデータ管理ができます。"
))

Pydanticのインスタンスをそのまま渡せます。 辞書への変換も、SQL文の組み立ても、一切不要です。

③ クエリ・全文検索

# Djangoライクなクエリ
articles = db.query(Article, author="neko", views__gte=0, order_by="id", desc=True)

# FTS5全文検索
results = db.search(Article, "SQLite")
for hit in results:
    print(f"✨ {hit.title}")

db.close()

たったこれだけです。 SQLを1行も書かずに、型安全なデータ管理・全文検索が実現できました。

「本当に使えるの？」を検証する

ここからが本番です。個人開発レベルで本当に実用に耐えるか、いくつかの角度から検証してみます。

バッチ挿入のパフォーマンス

ゲームのプレイヤーシステムを題材に、大量データを試してみました。

from datetime import datetime

class Player(BaseModel):
    __nyan_primary_key__ = "player_id"   # id以外を主キーにするときに指定
    player_id: int
    username: Indexed[str]
    level: Indexed[int]
    score: int = 0
    created_at: datetime

db = NyanSQLite("game.db")
db.register(Player)

players = [
    Player(player_id=i, username=f"player_{i}", level=i % 50, score=i * 100, created_at=datetime.now())
    for i in range(1, 100001)
]

import time
start = time.time()
db.insert_many(players)
print(f"Inserted in {time.time() - start:.4f}s")
# 結果: 0.6166s（チューニング後の実測）

10万件が約0.6秒。 最初の計測では1.2996sでしたが、チューニングで半分以下に短縮できました。内部でWALモード・トランザクション・自動チャンキング（32766プレースホルダ上限の自動分割）を組み合わせており、小〜中規模のデータなら十分すぎる速度です。

挿入後のクエリも直感的に書けます。

# ランキング上位10人
top = db.query(Player, order_by="score", desc=True, limit=10)

# レベル40以上を絞り込み
high_level = db.query(Player, level__gte=40, limit=100)

# 文字列形式のフィルタも使える（複雑な条件に）
mid_level = db.query(Player, "level > 10", "level < 40")

# 1件更新
db.update(Player, where={"player_id": 1}, score=9999)

# カウント
total = db.count(Player)
active = db.count(Player, level__gte=30)

クエリ演算子の全貌

使える演算子をまとめておきます。

db.query(Article, author="neko")                # 完全一致
db.query(Article, views__gt=10)                 # >
db.query(Article, views__gte=10)                # >=
db.query(Article, views__lt=100)                # <
db.query(Article, views__lte=100)               # <=
db.query(Article, views__ne=50)                 # !=
db.query(Article, title__like="%Python%")       # LIKE
db.query(Article, id__in=[1, 2, 3])            # IN
db.query(Article, author__is_null=False)        # NULL チェック

「Djangoのクエリセットほぼそのままじゃん」と感じた方、正解です。意図的にそう設計されています。

非同期（asyncio）対応

FastAPIと組み合わせることを想定して、非同期版も試してみました。

import asyncio
from nyansqlite import NyanSQLiteAIO

async def main():
    async with NyanSQLiteAIO("async.db") as db:
        db.register(Article)

        await db.insert(Article(id=1, author="alice", title="Hello", body="World"))

        articles = await db.query(Article, author="alice")
        print(articles[0].title)  # Hello

        await db.insert_many([
            Article(id=i, author="bob", title=f"記事{i}", body="...")
            for i in range(2, 10)
        ])

        count = await db.count(Article)
        print(f"合計: {count}件")

asyncio.run(main())

内部で asyncio.to_thread を使っているため、DB操作がブロッキングにならず、他の非同期タスクの妨げになりません。FastAPIのエンドポイントからそのまま呼び出せます。

「本番で使う」ときに知っておくべきこと

ここまで読んで「使ってみたい！」と思った方に、本番投入前に押さえておきたいポイントを共有します。

strict_deserialization：データ破損との戦い方

DBにJSON形式で保存されたカラムが壊れていた場合の挙動を2モードから選べます。

本番では strict_deserialization の選択を明示的に行ってください。

モード 動作 向いているケース
False（デフォルト） 警告を出して生データを返す ログ解析など、一部破損でも処理継続したい場合
True ValueError を即座に発生 金融・決済など、データ整合性が最重要の場合 
# 厳格モード（データ破損を即検出）
db = NyanSQLite("app.db", strict_deserialization=True)

try:
    articles = db.query(Article)
except ValueError as e:
    print(f"データ破損を検出: {e}")
    # ここで通知・ロールバック等の処理を行う

QueryValidationError：型ミスをランタイムで捕まえる

型が不正なクエリパラメータは QueryValidationError として検出されます。 SQLに渡る前に弾いてくれるのでインジェクション対策としても安心です。

from nyansqlite import QueryValidationError

try:
    db.query(Article, views__gt="数字じゃない")  # 型エラー
except QueryValidationError as e:
    print(f"不正なクエリパラメータ: {e}")

テーブル名衝突の検知

CamelCaseの正規化でテーブル名が衝突するケースを自動で検出してくれます。

from nyansqlite import TableNameCollisionError

class UserAuth(BaseModel):
    id: int

class User_Auth(BaseModel):    # 正規化後、どちらも "user_auth" になる
    id: int

db.register(UserAuth)
try:
    db.register(User_Auth)    # TableNameCollisionError が発生
except TableNameCollisionError as e:
    print(f"テーブル名の衝突を検出: {e}")

マルチスレッドも安全

すべてのDB操作が threading.Lock で保護されています。

import threading

db = NyanSQLite("thread_safe.db")
db.register(Article)

def worker():
    db.query(Article, author="neko")
    db.insert(Article(id=99, author="neko", title="スレッドテスト", body="..."))

threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()

高度な機能：使いこなすと差がつく

複合インデックス

from nyansqlite import CompositeIndex

class Order(BaseModel):
    __nyan_indexes__ = [
        CompositeIndex("user_id", "created_at"),
        CompositeIndex("product_id", "status", unique=True),
    ]
    id: int
    user_id: int
    product_id: int
    status: str

コンテキストマネージャー（書き忘れを防ぐ）

with NyanSQLite("app.db") as db:
    db.register(Article)
    db.insert(Article(...))
    # ブロックを抜けると自動で close() される

その他の便利メソッド

# 存在確認（count()より意図が明確）
if db.exists(Article, id=1):
    print("Found!")

# 特定カラムのみ取得（帯域節約・高速化）
titles = db.select(Article, ["title", "author"], views__gte=100)

# DB最適化（デプロイ前後に一度）
db.vacuum()

# FTSインデックス再構築（データが大量に変化した後に）
db.rebuild_fts(Article)

総評：正直に言います

好きなところ

  • Pydanticモデルが型定義とDBスキーマを兼ねるため、コードが1箇所にまとまる。「モデルを変えたらマイグレーションも忘れず直さないと……」という緊張感がない
  • Indexed[T] / Searchable[str] のアノテーション設計がエレガント。型ヒントがそのままインフラの設定になる感覚は、他のORMではなかなか味わえない
  • FTS5全文検索がゼロ設定で使えるのは地味に大きい。Meilisearchを別途立てなくていい
  • バッチ挿入の速度は想定以上。10万件0.3秒は素直に驚いた

気をつけるべきところ

  • Pydantic v2 が必須です。 v1系の既存プロジェクトへの導入はマイグレーションが先決です
  • SQLiteの性質上、数千万件超の大規模データ・複雑なJOINが必要なユースケースには不向きです。そういった用途には PostgreSQL + SQLAlchemy を
  • id フィールドが自動で主キーになる仕様のため、フィールド名が id でない場合は __nyan_primary_key__ の指定を必ず行ってください（忘れると主キーなしテーブルになります）

おわりに

NyanSQLiteは「大規模システムのORM」ではなく、「Pythonエンジニアが日常的に書く小〜中規模のデータ管理を、型安全かつ快適に」 するためのツールです。

こんなシーンにぴったりです！

  • ✅ CLIツール・スクリプトのデータ永続化
  • ✅ FastAPI / Starlette のローカルDB
  • ✅ プロトタイプ・個人開発・社内ツール
  • ✅ すでにPydanticを使っているプロジェクトへの自然な追加

Pythonを書くときの「型が消えていく不安」を、DBアクセスの層まで持ち込まずに済む体験は一度味わうと手放せません。ぜひ試してみてください🐱

pip install nyansqlite
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?