NanaSQLite チュートリアル
基礎から応用まで、NanaSQLiteを段階的に学ぶガイドです。
前提条件
- Python 3.9以上
- Pythonの辞書の基本的な理解
- SQLiteの知識があると役立ちますが、必須ではありません
インストール
# 基本インストール
pip install nanasqlite
# 推奨:高速化オプション付き(Zsh等のシェルではクォート推奨)
pip install "nanasqlite[speed]"学習パス(基礎〜応用)
レッスン1: 最初のデータベース
データベースの作成
from nanasqlite import NanaSQLite
# データベースファイルを作成または開く
db = NanaSQLite("tutorial.db")
# データを保存
db["greeting"] = "こんにちは、世界!"
db["number"] = 42
db["pi"] = 3.14159
# データを取得
print(db["greeting"]) # こんにちは、世界!
print(db["number"]) # 42
# 終了時にクローズ
db.close()何が起こったか?
- SQLiteデータベースファイル
tutorial.dbが作成されました - データは即座にディスクに保存されました
- プログラム終了後もデータは永続化されています
コンテキストマネージャの使用
from nanasqlite import NanaSQLite
# 自動的にデータベースをクローズ
with NanaSQLite("tutorial.db") as db:
db["message"] = "コンテキストマネージャを使用!"
print(db["message"])
# ここでデータベースは自動的にクローズされますベストプラクティス: 常にコンテキストマネージャ(with文)を使用して、適切なクリーンアップを保証しましょう。
レッスン2: 複雑なデータの扱い
ネスト構造の保存
with NanaSQLite("tutorial.db") as db:
# ユーザープロファイルを保存
db["user_alice"] = {
"name": "Alice",
"age": 30,
"email": "[email protected]",
"preferences": {
"theme": "dark",
"notifications": True,
"language": "ja"
},
"tags": ["admin", "developer", "python"]
}
# ネストしたデータにアクセス
user = db["user_alice"]
print(user["name"]) # Alice
print(user["preferences"]["theme"]) # dark
print(user["tags"][0]) # adminサポートされるデータ型
with NanaSQLite("tutorial.db") as db:
db["string"] = "テキスト"
db["integer"] = 100
db["float"] = 99.99
db["boolean"] = True
db["none"] = None
db["list"] = [1, 2, 3, "four"]
db["dict"] = {"nested": {"deeply": {"value": 123}}}注意
NanaSQLiteは複雑なPythonオブジェクトを自動的にJSONにシリアライズします。
レッスン3: 辞書操作
存在確認
with NanaSQLite("tutorial.db") as db:
db["config"] = {"theme": "dark"}
# キーの存在を確認
if "config" in db:
print("設定が存在します!")
if "missing" not in db:
print("このキーは存在しません")
# デフォルト値付きで取得
value = db.get("missing", "デフォルト値")
print(value) # デフォルト値データの反復処理
with NanaSQLite("tutorial.db") as db:
# データを追加
db["user_1"] = {"name": "Alice"}
db["user_2"] = {"name": "Bob"}
db["user_3"] = {"name": "Charlie"}
# キーを反復
for key in db.keys():
print(key)
# 値を反復
for value in db.values():
print(value)
# キーと値のペアを反復
for key, value in db.items():
print(f"{key}: {value}")更新と削除
with NanaSQLite("tutorial.db") as db:
# 単一キーの更新
db["counter"] = 0
db["counter"] = db["counter"] + 1
print(db["counter"]) # 1
# 複数キーを一度に更新
db.update({
"key1": "value1",
"key2": "value2",
"key3": "value3"
})
# キーを削除
del db["key1"]
# Pop(取得して削除)
value = db.pop("key2")
print(value) # value2
# 全データをクリア
# db.clear() # コメントを外すとすべて削除されますレッスン4: パフォーマンス最適化
一括ロード
# 読み込み重視のワークロードには、起動時に全データをロード
with NanaSQLite("tutorial.db", bulk_load=True) as db:
# 全データがメモリに展開されています
# 以降の読み込みは超高速
for key in db.keys():
print(db[key]) # データベースクエリなし!bulk_loadを使用するとき:
- 小~中規模のデータベース(<100MB)
- ほとんどのキーを頻繁に読み込む
- アプリケーション起動時間が重要でない
bulk_loadを使用しないとき:
- 大規模データベース(>1GB)
- まばらなアクセスパターン(少数のキーのみアクセス)
- メモリに制約がある環境
バッチ操作
with NanaSQLite("tutorial.db") as db:
# ❌ 遅い: 個別の挿入
for i in range(1000):
db[f"item_{i}"] = {"value": i}
# ✅ 速い: バッチ挿入(10-100倍高速)
data = {f"item_{i}": {"value": i} for i in range(1000)}
db.batch_update(data)
# バッチ削除
keys_to_delete = [f"item_{i}" for i in range(500)]
db.batch_delete(keys_to_delete)パフォーマンスのヒント: 100以上の操作には、常にバッチメソッドを使用しましょう。
レッスン5: Pydanticモデルの使用
from pydantic import BaseModel
from nanasqlite import NanaSQLite
class User(BaseModel):
name: str
age: int
email: str
with NanaSQLite("tutorial.db") as db:
# Pydanticモデルを保存
user = User(name="Alice", age=30, email="[email protected]")
db.set_model("user_alice", user)
# Pydanticモデルとして取得
retrieved = db.get_model("user_alice", User)
print(retrieved.name) # Alice
print(retrieved.age) # 30
print(type(retrieved)) # <class '__main__.User'>レッスン6: 直接SQLクエリ
基本的なクエリ
with NanaSQLite("tutorial.db") as db:
# カスタムテーブルを作成
db.create_table("users", {
"id": "INTEGER PRIMARY KEY",
"name": "TEXT NOT NULL",
"email": "TEXT UNIQUE",
"age": "INTEGER"
})
# データを挿入
db.sql_insert("users", {"name": "Alice", "email": "[email protected]", "age": 30})
db.sql_insert("users", {"name": "Bob", "email": "[email protected]", "age": 25})
# データをクエリ
results = db.query(
table_name="users",
columns=["name", "age"],
where="age > ?",
parameters=(25,),
order_by="name ASC"
)
for row in results:
print(f"{row['name']}: {row['age']}")高度なSQL
with NanaSQLite("tutorial.db") as db:
# カスタムSQLを実行
cursor = db.execute("SELECT * FROM users WHERE name LIKE ?", ("A%",))
for row in cursor:
print(row)
# 全結果を取得
rows = db.fetch_all("SELECT name, age FROM users ORDER BY age DESC")
# 1つの結果を取得
row = db.fetch_one("SELECT * FROM users WHERE email = ?", ("[email protected]",))レッスン7: エラーハンドリング
from nanasqlite import NanaSQLite
with NanaSQLite("tutorial.db") as db:
# 存在しないキーを処理
try:
value = db["nonexistent"]
except KeyError:
print("キーが見つかりません!")
# より良い方法: デフォルト値付きでget()を使用
value = db.get("nonexistent", "デフォルト")
# SQLエラーを処理
import apsw
try:
db.execute("INVALID SQL")
except apsw.Error as e:
print(f"SQLエラー: {e}")レッスン8: 複数テーブル
ひとつのデータベースファイル内で、異なるデータタイプごとにテーブルを分けることができます。.table() メソッドを使用すると、同じ接続を共有しながら独立したテーブルを操作できます。
from nanasqlite import NanaSQLite
# メインインスタンスの作成
db = NanaSQLite("app.db")
# サブテーブル用インスタンスの取得(接続を共有するため効率的)
users_db = db.table("users")
config_db = db.table("config")
cache_db = db.table("cache")
# それぞれが独立して動作
users_db["alice"] = {"name": "Alice", "role": "admin"}
config_db["theme"] = "dark"
cache_db["temp_data"] = {"expires": "2024-12-31"}
# 親を閉じればすべてクローズされます
db.close()レッスン9: 非同期の使用(上級)
FastAPIなどの非同期フレームワーク向け:
import asyncio
from nanasqlite import AsyncNanaSQLite
async def main():
async with AsyncNanaSQLite("tutorial.db") as db:
# 非同期操作
await db.aset("user", {"name": "Alice"})
user = await db.aget("user")
print(user)
# 並行操作
results = await asyncio.gather(
db.aget("key1"),
db.aget("key2"),
db.aget("key3")
)
asyncio.run(main())詳細な非同期ドキュメントはasync_guideを参照してください。
機能別レッスン(v1.3+)
レッスン 10: キャッシュ戦略 (v1.3.0)
メモリ使用量を抑えつつ高速に動作させるために、キャッシュの追い出し(LRU)戦略を選択できます。
LRU キャッシュの使用
from nanasqlite import NanaSQLite, CacheType
# 最新の 1000 件のみをメモリにキャッシュ
with NanaSQLite("app.db", cache_strategy=CacheType.LRU, cache_size=1000) as db:
db["key"] = "value"TTL キャッシュの使用 (v1.3.1)
有効期限のあるデータ(セッション、一時的なキャッシュなど)を扱う場合に便利です。
# 1時間 (3600秒) で失効するキャッシュ
# cache_persistence_ttl=True を設定すると、失効時に SQLite からも自動削除されます
with NanaSQLite("app.db",
cache_strategy=CacheType.TTL,
cache_ttl=3600,
cache_persistence_ttl=True
) as db:
db["session_123"] = {"user_id": 42}パフォーマンスの最大化
orjson と lru-dict を導入することで、シリアライズとキャッシュ操作の両方を劇的に高速化できます。
pip install "nanasqlite[speed]"詳細はパフォーマンスチューニングガイドを参照してください。
レッスン 11: 暗号化 (v1.3.1)
機密データを安全に保存するために、透過的な暗号化をサポートしています。
基本的な暗号化
from nanasqlite import NanaSQLite
# 32バイトのキーを指定(AES-GCM がデフォルトで使用されます)
db = NanaSQLite("secure.db", encryption_key=b"your-32-byte-secure-key-here-!!!")
db["secret"] = {"password": "top-secret-password"}
print(db["secret"]) # 通常通りアクセス可能暗号化方式の選択
環境に応じて最適なモードを選択できます:
aes-gcm(標準): 高速かつ省サイズ。chacha20: ハードウェア加速がない環境で高速。fernet: 従来の互換性重視。
db = NanaSQLite("secure.db",
encryption_key=key,
encryption_mode="chacha20" # "aes-gcm", "chacha20", "fernet" から選択
)暗号化モードの選び方
- AES-GCM (デフォルト):
cryptographyが推奨する最も安全で標準的な方式です。ハードウェア加速(AES-NI)がある環境(現代的なPCやサーバー)で非常に高速です。 - ChaCha20-Poly1305: ハードウェア加速がない環境(低電力なARMデバイスなど)でも高速に動作するソフトウェア実装の暗号化です。
- Fernet: 鍵管理が最もシンプルで、従来バージョンとの互換性や使いやすさを重視する場合に適しています。
ハイブリッド設計: データはSQLite上では暗号化されますが、メモリキャッシュ内は平文で保持されるため、読み取り速度を犠牲にすることなく安全性を確保できます。
レッスン 12: インストールオプション
NanaSQLite は、用途に合わせて必要な依存関係のみをインストールできるように「Extra」機能を提供しています。
| オプション名 | 内容 | 主な用途 |
|---|---|---|
[speed] | orjson, lru-dict | パフォーマンスを最大化したい場合 |
[encryption] | cryptography | 暗号化機能を使用したい場合 |
[all] | 上記すべて | すべての機能を利用したい場合 |
[dev] | pytest, ruff, mypy 等 | NanaSQLite自体の開発やテストを行う場合 |
# Zsh などのシェルでは角括弧を解釈させないためにクォートが必要です
pip install "nanasqlite[all]"レッスン13: ロックタイムアウト・バックアップ・リストア (v1.3.4b1以降)
ロックタイムアウト
デフォルトでは NanaSQLite は内部ロックを無制限に待ち続けます。 lock_timeout を設定することで、ロックが一定時間内に取得できない場合に NanaSQLiteLockError を送出できます:
from nanasqlite import NanaSQLite, NanaSQLiteLockError
db = NanaSQLite("app.db", lock_timeout=2.0) # 2秒待ってもロックが取れなければエラー
try:
db["key"] = "value"
except NanaSQLiteLockError as e:
print(f"ロックを取得できませんでした: {e}")lock_timeout を使う場面:
- マルチスレッドのアプリケーションでデッドロックが起こりうる場合
- 応答時間を上限で保証したいサービス
バックアップ
backup() は APSW の SQLite オンラインバックアップ API を使用して、データベースを一貫性のある状態で別ファイルにコピーします。 SQLite のトランザクション機構により「データベースが使用中」であっても壊れることなく安全に実行できます。 バックアップ実行中に NanaSQLite の内部ロックを保持しないため、同一プロセス内の他の NanaSQLite 操作をブロックしません:
db = NanaSQLite("app.db")
db["user"] = {"name": "Nana", "role": "admin"}
# バックアップを作成 — ノンブロッキング: バックアップ中も他の NanaSQLite 操作は通常通り動作します
db.backup("app_backup_2026-03-04.db")
# バックアップファイルは完全に独立した SQLite データベース
backup_db = NanaSQLite("app_backup_2026-03-04.db")
print(backup_db["user"]) # {'name': 'Nana', 'role': 'admin'}
backup_db.close()リストア
restore() は現在のデータベースをバックアップファイルで置き換え、自動的に再接続します:
db = NanaSQLite("app.db")
db["counter"] = 1
# ベースラインスナップショットを作成
db.backup("snapshot.db")
# データの誤操作や破損をシミュレート
db["counter"] = 9999
db["bad_key"] = "oops"
# スナップショットまでロールバック
db.restore("snapshot.db")
print(db["counter"]) # 1
print("bad_key" in db) # False
db.close()NOTE
restore() は プライマリ(接続を所有している)インスタンスからのみ呼び出せます。 .table() で取得したインスタンスからは呼び出せません。 また、リストア後はメモリキャッシュが自動的にクリアされます。
実践リファレンス
一般的なパターン
設定の保存
with NanaSQLite("config.db") as db:
# アプリケーション設定を保存
db["app_config"] = {
"version": "1.0.0",
"debug": False,
"database_url": "sqlite:///data.db",
"secret_key": "your-secret-key"
}
# 設定を取得
config = db["app_config"]
if config["debug"]:
print("デバッグモード有効")キャッシング
import time
with NanaSQLite("cache.db") as db:
# タイムスタンプ付きでキャッシュデータを保存
db["api_response"] = {
"data": {"users": [...]},
"cached_at": time.time()
}
# キャッシュの年齢を確認
cached = db.get("api_response")
if cached and (time.time() - cached["cached_at"]) < 3600:
# キャッシュが新鮮(1時間未満)
data = cached["data"]
else:
# APIから新しいデータを取得
# 例: data = requests.get("https://api.example.com/data").json()
data = fetch_from_api() # 実際のAPIコール関数に置き換えてください
db["api_response"] = {"data": data, "cached_at": time.time()}セッション保存
import uuid
import time
with NanaSQLite("sessions.db") as db:
# セッションを作成
session_id = str(uuid.uuid4())
db[f"session_{session_id}"] = {
"user_id": "alice",
"created_at": time.time(),
"data": {"cart": ["item1", "item2"]}
}
# セッションを取得
session = db.get(f"session_{session_id}")
if session:
print(f"ユーザー: {session['user_id']}")
print(f"カート: {session['data']['cart']}")ベストプラクティス
- 常にコンテキストマネージャを使用(
with文) - 複数書き込みにはバッチ操作を使用(100以上のアイテム)
- 読み込み重視のワークロードには
bulk_load=Trueを使用 try/except KeyErrorの代わりにデフォルト値付きでget()を使用- 異なるテーブルで関心事を分離
- 終了時にデータベースをクローズ(コンテキストマネージャで自動)
次のステップ
- 完全なメソッドドキュメントはAPIリファレンスを参照
- async/await の使用方法はAsync Guideを探索
- 本番環境のヒントはBest Practicesを確認
- 実際のコードサンプルはexamples/を参照
トラブルシューティング
データベースがロックされている
# SQLiteは一度に1つの書き込みのみ許可
# 複数書き込みにはトランザクションを使用
with db.transaction():
db["key1"] = "value1"
db["key2"] = "value2"メモリ使用量が多い
# 大規模データベースにはbulk_loadを使用しない
# デフォルトの遅延ロードを使用
db = NanaSQLite("large.db", bulk_load=False)パフォーマンスが遅い
# 一括書き込みにはバッチ操作を使用
data = {f"key_{i}": value for i in range(10000)}
db.batch_update(data) # 個別書き込みよりはるかに高速まとめ
学習内容:
- ✅ 基本的なCRUD操作
- ✅ 複雑なネストデータの扱い
- ✅ パフォーマンス最適化(bulk_load、バッチ操作)
- ✅ Pydantic統合
- ✅ 直接SQLクエリ
- ✅ エラーハンドリング
- ✅ 一般的な使用パターン
- ✅ キャッシュ戦略(LRU/TTL)
- ✅ 暗号化とモード選択
- ✅ 機能別インストールオプション
- ✅ ロックタイムアウト・バックアップ・リストアによる安全運用
NanaSQLiteを楽しんでコーディングしてください!