この記事では、SQLiteをPythonの辞書のように直感的に扱えるライブラリ「DictSQLite」の開発背景、技術的な工夫、そして現在開発中のv2.xの新機能について紹介します。
はじめに
こんにちは!高校生エンジニアの@harumaki4649です。普段はDiscord botのDisnanaを運営しています。
今回は、SQLiteデータベースをPythonの辞書のように直感的に扱えるライブラリ「DictSQLite」を開発したので、その設計思想や実装のポイント、そして現在開発中のv2.xの新機能を紹介します。
pip install dictsqlite
from dictsqlite import DictSQLite
# 辞書のように使える!
db = DictSQLite('sample.db')
db['name'] = 'Alice'
db['age'] = 30
print(db['name']) # Alice
print('age' in db) # True
# テーブルも簡単
users = db.table('users')
users['user1'] = {'name': 'Bob', 'age': 25}
なぜ作ったのか?
課題認識
SQLiteは軽量で組み込みが簡単なデータベースですが、シンプルなKey-Value操作にも以下の問題がありました:
| 課題 | 具体例 |
|---|---|
| SQL記述の手間 | 単純なCRUD操作でも毎回SQL文を書く必要がある |
| スキーマ管理 | テーブル存在確認、カラム追加などが煩雑 |
| 直感性の欠如 | Pythonの辞書のような直感的なインターフェースがない |
| 並行制御 | マルチプロセス/スレッド環境での競合制御が難しい |
実際にDisnanaのDiscord bot開発で、設定データを保存するたびに以下のようなコードを書いていました:
従来の実装例(クリックして展開)
# 従来のSQLite操作
import sqlite3
conn = sqlite3.connect('config.db')
cursor = conn.cursor()
# テーブル作成
cursor.execute('''
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT
)
''')
# データ挿入
cursor.execute('INSERT OR REPLACE INTO config VALUES (?, ?)', ('theme', 'dark'))
conn.commit()
# データ取得
cursor.execute('SELECT value FROM config WHERE key = ?', ('theme',))
result = cursor.fetchone()
theme = result[0] if result else None
conn.close()
そこで、「SQLiteを辞書として扱えたら便利では?」というコンセプトで開発したのがDictSQLiteです。
主な機能(v1.x系)
1. 辞書ライクなAPI
Pythonの辞書と同じ感覚でデータベースを操作できます:
from dictsqlite import DictSQLite
db = DictSQLite('sample.db')
# 書き込み
db['name'] = 'Alice'
db['age'] = 30
db.update({'city': 'Tokyo', 'country': 'Japan'})
# 読み込み
print(db['name']) # Alice
print(db.get('city')) # Tokyo
print(db.get('missing')) # None
# 削除
del db['country']
# 存在確認
print('name' in db) # True
print('country' in db) # False
# イテレーション
for key, value in db.items():
print(f"{key}: {value}")
db.close()
コンテキストマネージャーも使えます!
with DictSQLite('app.db') as db:
db['status'] = 'active'
2. テーブル機能
複数のテーブルを名前空間として使い分けることも可能です:
db = DictSQLite('myapp.db')
# ユーザーテーブル
users = db.table('users')
users['user1'] = {'name': 'Bob', 'age': 25}
users['user2'] = {'name': 'Charlie', 'age': 35}
# 設定テーブル
config = db.table('config')
config['theme'] = 'dark'
config['language'] = 'ja'
print(users['user1']) # {'name': 'Bob', 'age': 25}
print(config['theme']) # dark
3. トランザクション管理
コンテキストマネージャーで簡単にトランザクションを扱えます:
db = DictSQLite('app.db')
try:
with db.transaction() as t:
t['status'] = 'pending'
t['count'] = 100
# エラーが発生した場合、自動的にロールバック
raise ValueError("Something went wrong")
except ValueError:
pass
# トランザクションがロールバックされている
print(db.get('status')) # None
4. 組み込み暗号化
機密データを保存する場合は、暗号化オプションを使用してください!
# AES暗号化を有効化
db = DictSQLite('encrypted.db', encryption_key='my_secret_key_12345678')
db['password'] = 'super_secret_password'
db['api_key'] = 'sk-xxxxxxxxxxxxx'
# データベースファイルは暗号化されて保存される
db.close()
5. プロセス/スレッドセーフ
portalockerを使用して、複数プロセスからの同時アクセスでもデータ整合性を保証します。
# 複数のプロセスから同時にアクセスしても安全
from multiprocessing import Process
def worker(worker_id):
db = DictSQLite('shared.db')
db[f'worker_{worker_id}'] = f'Data from worker {worker_id}'
db.close()
processes = [Process(target=worker, args=(i,)) for i in range(10)]
for p in processes:
p.start()
for p in processes:
p.join()
v2.x系の新機能(開発中)
v2.x系は現在開発中です! 以下の機能は将来のバージョンで利用可能になります。
現在開発中のv2.x系では、パフォーマンスと機能性が大幅に向上します:
非同期対応(Async/Await)
from dictsqlite import AsyncDictSQLite
async def main():
db = AsyncDictSQLite(':memory:')
# 非同期API
await db.aset('key', 'value')
value = await db.aget('key')
# バッチ操作も高速
await db.abatch_set([
('key1', 'value1'),
('key2', 'value2'),
('key3', 'value3'),
])
await db.close()
ホットキャッシュ機能
v2では、頻繁にアクセスされるデータをメモリ上にキャッシュする「ホットティア」を実装:
db = DictSQLite(
'app.db',
hot_capacity=1_000_000, # ホットキャッシュの最大サイズ
persist_mode='lazy' # 遅延書き込みで高速化
)
# 頻繁にアクセスされるデータは自動的にキャッシュされる
for i in range(10000):
db[f'key_{i}'] = f'value_{i}' # 超高速!
JSON Binary(JSONB)サポート
# JSONBモードで保存
db = DictSQLite('data.db', storage_mode='jsonb')
db['user'] = {
'name': 'Alice',
'age': 30,
'tags': ['developer', 'python', 'sqlite']
}
# Pickleよりも互換性が高く、他言語からも読める
Safe Pickle機能
信頼できないデータをunpickleする際のセキュリティ対策!
db = DictSQLite(
'safe.db',
enable_safe_pickle=True,
safe_pickle_allowed_modules=['myapp.models']
)
# 許可されたモジュールのオブジェクトのみunpickle可能
パフォーマンス統計
db = DictSQLite('perf.db', hot_capacity=10000)
# 大量データ挿入
for i in range(100000):
db[f'key_{i}'] = f'value_{i}'
# 統計情報を確認
stats = db.stats()
print(stats)
# {
# 'encryption_enabled': False,
# 'hot_tier_size': 10000,
# 'persist_mode': 'writethrough',
# 'storage_mode': 'pickle',
# ...
# }
技術的な工夫
自動スキーマ管理
DictSQLiteは初回アクセス時に自動的にテーブルとカラムを作成します。
内部では以下のような構造を使用:
CREATE TABLE IF NOT EXISTS main (
key TEXT PRIMARY KEY,
value BLOB -- Pickleシリアライズされた値
);
ユーザーはスキーマを意識せずにデータを格納できます。
型の柔軟性
Pythonの任意の型(dict, list, カスタムオブジェクトなど)を保存できます:
db = DictSQLite('flexible.db')
# 様々な型を保存可能
db['string'] = 'Hello, World!'
db['number'] = 42
db['list'] = [1, 2, 3, 4, 5]
db['dict'] = {'nested': {'data': 'value'}}
db['custom'] = MyCustomClass(param=123)
# すべて正しく復元される
assert isinstance(db['list'], list)
assert isinstance(db['custom'], MyCustomClass)
GitHub Copilotとの協働開発
最近のコミット履歴を見ていただくとわかる通り、GitHub Copilot SWE Agentを積極的に活用しています:
# 最近のコミット例
✅ Fix UnicodeEncodeError in build.sh Python print statements
✅ Fix safe_pickle import fallback and remove dead code
✅ Add comprehensive documentation and test coverage
✅ Add __repr__ to TableProxy for meaningful print output
人間とAIの協働開発により、以下を実現:
- バグ修正の自動化
- テストカバレッジの向上
- コード品質の改善
パフォーマンス
ベンチマークスクリプトを用意しており、詳細はbenchmarkディレクトリをご覧ください。
ベンチマーク結果が更新されていな場合がございます
ユースケース
1. Discord Botの設定管理
from dictsqlite import DictSQLite
# ギルドごとの設定を保存
class BotConfig:
def __init__(self):
self.db = DictSQLite('bot_config.db')
self.guilds = self.db.table('guilds')
def set_prefix(self, guild_id, prefix):
config = self.guilds.get(str(guild_id), {})
config['prefix'] = prefix
self.guilds[str(guild_id)] = config
def get_prefix(self, guild_id):
config = self.guilds.get(str(guild_id), {})
return config.get('prefix', '!')
2. 軽量キャッシュシステム
import requests
from dictsqlite import DictSQLite
import time
class APICache:
def __init__(self, ttl=3600):
self.cache = DictSQLite('api_cache.db')
self.ttl = ttl
def get(self, url):
cached = self.cache.get(url)
if cached and time.time() - cached['timestamp'] < self.ttl:
return cached['data']
# キャッシュミス
response = requests.get(url).json()
self.cache[url] = {
'timestamp': time.time(),
'data': response
}
return response
3. 設定ファイルの永続化
from dictsqlite import DictSQLite
class AppSettings:
def __init__(self):
self.settings = DictSQLite('settings.db')
self._load_defaults()
def _load_defaults(self):
defaults = {
'theme': 'dark',
'language': 'ja',
'notifications': True
}
for key, value in defaults.items():
if key not in self.settings:
self.settings[key] = value
def __getitem__(self, key):
return self.settings[key]
def __setitem__(self, key, value):
self.settings[key] = value
インストールと基本的な使い方
インストール
pip install dictsqlite
クイックスタート
from dictsqlite import DictSQLite
# データベース作成
db = DictSQLite('myapp.db')
# データ操作
db['username'] = 'alice'
db['score'] = 100
# データ取得
print(db['username']) # alice
# 存在確認
if 'score' in db:
db['score'] += 10
# イテレーション
for key, value in db.items():
print(f"{key} = {value}")
# クローズ
db.close()
詳細なドキュメントはGitHubリポジトリを参照してください!
まとめ
DictSQLiteは「シンプルさ」と「機能性」のバランスを重視して設計したライブラリです。SQLiteの堅牢性と辞書の直感性を組み合わせることで、開発者の生産性を向上させることを目指しています。
今後の展望
- ✅ v2.xの安定版リリース(非同期対応、パフォーマンス改善)
- 🔄 クエリビルダーの実装
- 🔄 マルチテーブルJOIN対応
- 🔄 インデックス最適化
現在も活発に開発を続けており、貢献やフィードバックをお待ちしています!
リンク集
もしよければ⭐スターやコントリビューションをお待ちしています!質問やフィードバックはIssuesまたはsupport@disnana.comまでお気軽にどうぞ!