nanasqlite 1.1.0a4

A dict-like SQLite wrapper with APSW for instant persistence and memory caching

NanaSQLite

A dict-like SQLite wrapper with instant persistence and intelligent caching.

English | 日本語

---

English

🚀 Features

• Dict-like Interface: Use familiar db["key"] = value syntax
• Instant Persistence: All writes are immediately saved to SQLite
• Smart Caching: Lazy load (on-access) or bulk load (all at once)
• Nested Structures: Full support for nested dicts and lists (up to 30+ levels)
• High Performance: WAL mode, mmap, and batch operations for maximum speed
• Zero Configuration: Works out of the box with sensible defaults

📦 Installation

pip install nanasqlite

⚡ Quick Start

from nanasqlite import NanaSQLite

# Create or open a database
db = NanaSQLite("mydata.db")

# Use it like a dict
db["user"] = {"name": "Nana", "age": 20, "tags": ["admin", "active"]}
print(db["user"])  # {'name': 'Nana', 'age': 20, 'tags': ['admin', 'active']}

# Data persists automatically
db.close()

# Reopen later - data is still there!
db = NanaSQLite("mydata.db")
print(db["user"]["name"])  # 'Nana'

🔧 Advanced Usage

# Bulk load for faster repeated access
db = NanaSQLite("mydata.db", bulk_load=True)

# Batch operations for high-speed writes
db.batch_update({
    "key1": "value1",
    "key2": "value2",
    "key3": {"nested": "data"}
})

# Context manager support
with NanaSQLite("mydata.db") as db:
    db["temp"] = "value"

📚 Documentation

• English Documentation
• API Reference

✨ New Features (v1.0.3rc3+)

Pydantic Support:

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

db.set_model("user", User(name="Nana", age=20))
user = db.get_model("user", User)

Direct SQL Execution:

# Execute custom SQL
cursor = db.execute("SELECT * FROM data WHERE key LIKE ?", ("user%",))
rows = db.fetch_all("SELECT key, value FROM data")

SQLite Wrapper Functions:

# Create tables and indexes easily
db.create_table("users", {
    "id": "INTEGER PRIMARY KEY",
    "name": "TEXT NOT NULL",
    "email": "TEXT UNIQUE"
})
db.create_index("idx_users_email", "users", ["email"])

# Simple queries
results = db.query(table_name="users", where="age > ?", parameters=(20,))

✨ Additional Features (v1.0.3rc4+)

22 new wrapper functions for comprehensive SQLite operations:

# Data operations
rowid = db.sql_insert("users", {"name": "Alice", "age": 25})
db.sql_update("users", {"age": 26}, "name = ?", ("Alice",))
db.upsert("users", {"id": 1, "name": "Alice", "age": 25})
total = db.count("users", "age >= ?", (18,))

# Query extensions (pagination, grouping)
page2 = db.query_with_pagination("users", limit=10, offset=10)
stats = db.query_with_pagination("orders", 
    columns=["user_id", "COUNT(*) as count"], group_by="user_id")

# Schema management
db.alter_table_add_column("users", "phone", "TEXT")
schema = db.get_table_schema("users")
db.drop_table("old_table", if_exists=True)

# Utilities & transactions
db.vacuum()  # Optimize database
with db.transaction():
    db.sql_insert("logs", {"message": "Event"})

✨ Multi-Table Support (v1.1.0dev1+)

Safely operate multiple tables in the same database with shared connections:

from nanasqlite import NanaSQLite

# Create main table instance
main_db = NanaSQLite("mydata.db", table="users")

# Get another table instance sharing the same connection
products_db = main_db.table("products")
orders_db = main_db.table("orders")

# Each table has isolated cache and operations
main_db["user1"] = {"name": "Alice", "email": "[e-mail]"}
products_db["prod1"] = {"name": "Laptop", "price": 999}
orders_db["order1"] = {"user": "user1", "product": "prod1"}

# Thread-safe concurrent writes to different tables
from concurrent.futures import ThreadPoolExecutor

def write_users(i):
    main_db[f"user{i}"] = {"name": f"User{i}"}

def write_products(i):
    products_db[f"prod{i}"] = {"name": f"Product{i}"}

with ThreadPoolExecutor(max_workers=10) as executor:
    executor.map(write_users, range(100))
    executor.map(write_products, range(100))

# Close only the main instance (closes shared connection)
main_db.close()

Key features:

• Shared connection & lock: All table instances share the same SQLite connection and thread lock
• Thread-safe: Concurrent writes to different tables are safely synchronized
• Memory efficient: Reuses connections instead of creating new ones
• Isolated cache: Each table maintains its own memory cache
• Works with async: await db.table("table_name") for AsyncNanaSQLite

⚠️ Important Usage Notes:

1. Do not create multiple instances for the same table:

   # ❌ BAD: Creates cache inconsistency
   users1 = db.table("users")
   users2 = db.table("users")  # Different cache, same DB table!
   
   # ✅ GOOD: Reuse the same instance
   users_db = db.table("users")
   # Use users_db throughout your code
   

   Each instance has its own independent cache. Multiple instances of the same table can lead to cache inconsistency at the memory level (though database writes remain correct).

2. Use context managers to avoid issues after close:

   # ✅ RECOMMENDED: Context manager ensures proper cleanup
   with NanaSQLite("app.db", table="main") as main_db:
       sub_db = main_db.table("sub")
       sub_db["key"] = "value"
   # Automatically closed, no orphaned instances
   
   # ❌ AVOID: Manual close can leave orphaned sub-instances
   main_db = NanaSQLite("app.db")
   sub_db = main_db.table("sub")
   main_db.close()  # sub_db may still access cached data
   

Best practices:

• Store table instances in variables and reuse them
• Prefer context managers (with statement) for automatic resource management
• Close the parent instance when done; child instances share the same connection

✨ Async Support (v1.0.3rc7+)

Full async/await support with optimized thread pool for high-performance non-blocking operations:

import asyncio
from nanasqlite import AsyncNanaSQLite

async def main():
    # Use async context manager with optimized thread pool
    async with AsyncNanaSQLite("mydata.db", max_workers=10) as db:
        # Async dict-like operations
        await db.aset("user", {"name": "Nana", "age": 20})
        user = await db.aget("user")
        print(user)  # {'name': 'Nana', 'age': 20}
        
        # Async batch operations
        await db.batch_update({
            "key1": "value1",
            "key2": "value2",
            "key3": {"nested": "data"}
        })
        
        # Concurrent operations (high-performance with thread pool)
        results = await asyncio.gather(
            db.aget("key1"),
            db.aget("key2"),
            db.aget("key3")
        )
        
        # Async SQL execution
        await db.create_table("users", {
            "id": "INTEGER PRIMARY KEY",
            "name": "TEXT",
            "age": "INTEGER"
        })
        await db.sql_insert("users", {"name": "Alice", "age": 25})
        users = await db.query("users", where="age > ?", parameters=(20,))
        
        # Multi-table support in async
        products_db = await db.table("products")
        await products_db.aset("prod1", {"name": "Laptop", "price": 999})

asyncio.run(main())

Performance optimizations:

• Dedicated thread pool executor (configurable with max_workers)
• APSW-based for maximum SQLite performance
• WAL mode and connection optimizations
• Ideal for high-concurrency scenarios

Perfect for async frameworks:

• FastAPI, Quart, Sanic (async web frameworks)
• aiohttp (async HTTP client/server)
• Discord.py, Telegram bots (async bots)
• Any asyncio-based application

---

日本語

🚀 特徴

• dict風インターフェース: おなじみの db["key"] = value 構文で操作
• 即時永続化: 書き込みは即座にSQLiteに保存
• スマートキャッシュ: 遅延ロード（アクセス時）または一括ロード（起動時）
• ネスト構造対応: 30階層以上のネストしたdict/listをサポート
• 高性能: WALモード、mmap、バッチ操作で最高速度を実現
• 設定不要: 合理的なデフォルト設定でそのまま動作

📦 インストール

pip install nanasqlite

⚡ クイックスタート

from nanasqlite import NanaSQLite

# データベースを作成または開く
db = NanaSQLite("mydata.db")

# dictのように使う
db["user"] = {"name": "Nana", "age": 20, "tags": ["admin", "active"]}
print(db["user"])  # {'name': 'Nana', 'age': 20, 'tags': ['admin', 'active']}

# データは自動的に永続化
db.close()

# 後で再度開いても、データはそのまま！
db = NanaSQLite("mydata.db")
print(db["user"]["name"])  # 'Nana'

🔧 高度な使い方

# 一括ロードで繰り返しアクセスを高速化
db = NanaSQLite("mydata.db", bulk_load=True)

# バッチ操作で高速書き込み
db.batch_update({
    "key1": "value1",
    "key2": "value2",
    "key3": {"nested": "data"}
})

# コンテキストマネージャ対応
with NanaSQLite("mydata.db") as db:
    db["temp"] = "value"

📚 ドキュメント

• 日本語ドキュメント
• APIリファレンス

✨ 新機能 (v1.0.3rc3+)

Pydantic互換性:

from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

db.set_model("user", User(name="Nana", age=20))
user = db.get_model("user", User)

直接SQL実行:

# カスタムSQLの実行
cursor = db.execute("SELECT * FROM data WHERE key LIKE ?", ("user%",))
rows = db.fetch_all("SELECT key, value FROM data")

SQLiteラッパー関数:

# テーブルとインデックスを簡単に作成
db.create_table("users", {
    "id": "INTEGER PRIMARY KEY",
    "name": "TEXT NOT NULL",
    "email": "TEXT UNIQUE"
})
db.create_index("idx_users_email", "users", ["email"])

# シンプルなクエリ
results = db.query(table_name="users", where="age > ?", parameters=(20,))

✨ マルチテーブルサポート (v1.1.0dev1+)

同一データベース内の複数テーブルを接続共有で安全に操作:

from nanasqlite import NanaSQLite

# メインテーブルインスタンスを作成
main_db = NanaSQLite("mydata.db", table="users")

# 同じ接続を共有する別のテーブルインスタンスを取得
products_db = main_db.table("products")
orders_db = main_db.table("orders")

# 各テーブルは独立したキャッシュと操作を持つ
main_db["user1"] = {"name": "Alice", "email": "[e-mail]"}
products_db["prod1"] = {"name": "Laptop", "price": 999}
orders_db["order1"] = {"user": "user1", "product": "prod1"}

# 異なるテーブルへのスレッドセーフな並行書き込み
from concurrent.futures import ThreadPoolExecutor

def write_users(i):
    main_db[f"user{i}"] = {"name": f"User{i}"}

def write_products(i):
    products_db[f"prod{i}"] = {"name": f"Product{i}"}

with ThreadPoolExecutor(max_workers=10) as executor:
    executor.map(write_users, range(100))
    executor.map(write_products, range(100))

# メインインスタンスのみをクローズ（共有接続を閉じる）
main_db.close()

主な特徴:

• 接続とロックの共有: 全てのテーブルインスタンスが同じSQLite接続とスレッドロックを共有
• スレッドセーフ: 異なるテーブルへの並行書き込みが安全に同期される
• メモリ効率: 新しい接続を作成せず、既存の接続を再利用
• キャッシュ分離: 各テーブルは独自のメモリキャッシュを保持
• 非同期対応: AsyncNanaSQLiteでは await db.table("table_name") で使用可能

⚠️ 重要な使用上の注意:

1. 同じテーブルに対して複数のインスタンスを作成しないでください:

   # ❌ 非推奨: キャッシュ不整合を引き起こす
   users1 = db.table("users")
   users2 = db.table("users")  # 異なるキャッシュ、同じDBテーブル！
   
   # ✅ 推奨: 同じインスタンスを再利用
   users_db = db.table("users")
   # コード全体でusers_dbを使用する
   

   各インスタンスは独立したキャッシュを持ちます。同じテーブルに対して複数のインスタンスを作成すると、メモリレベルでのキャッシュ不整合が発生する可能性があります（ただし、データベースへの書き込みは正しく行われます）。

2. close後の問題を避けるため、コンテキストマネージャを使用してください:

   # ✅ 推奨: コンテキストマネージャで適切にクリーンアップ
   with NanaSQLite("app.db", table="main") as main_db:
       sub_db = main_db.table("sub")
       sub_db["key"] = "value"
   # 自動的にクローズされ、孤立したインスタンスなし
   
   # ❌ 非推奨: 手動closeは孤立したサブインスタンスを残す可能性
   main_db = NanaSQLite("app.db")
   sub_db = main_db.table("sub")
   main_db.close()  # sub_dbはまだキャッシュデータにアクセスできる
   

ベストプラクティス:

• テーブルインスタンスを変数に保存して再利用する
• 自動リソース管理のためコンテキストマネージャ（with文）を優先する
• 完了時は親インスタンスをクローズする（子インスタンスは同じ接続を共有）

✨ 非同期サポート (v1.0.3rc7+)

高速化されたスレッドプールによる完全な async/await サポート:

import asyncio
from nanasqlite import AsyncNanaSQLite

async def main():
    # 最適化されたスレッドプールで非同期コンテキストマネージャを使用
    async with AsyncNanaSQLite("mydata.db", max_workers=10) as db:
        # 非同期dict風操作
        await db.aset("user", {"name": "Nana", "age": 20})
        user = await db.aget("user")
        print(user)  # {'name': 'Nana', 'age': 20}
        
        # 非同期バッチ操作
        await db.batch_update({
            "key1": "value1",
            "key2": "value2",
            "key3": {"nested": "data"}
        })
        
        # 並行操作（スレッドプールにより高性能）
        results = await asyncio.gather(
            db.aget("key1"),
            db.aget("key2"),
            db.aget("key3")
        )
        
        # 非同期SQL実行
        await db.create_table("users", {
            "id": "INTEGER PRIMARY KEY",
            "name": "TEXT",
            "age": "INTEGER"
        })
        await db.sql_insert("users", {"name": "Alice", "age": 25})
        users = await db.query("users", where="age > ?", parameters=(20,))
        
        # 非同期でのマルチテーブルサポート
        products_db = await db.table("products")
        await products_db.aset("prod1", {"name": "Laptop", "price": 999})

asyncio.run(main())

パフォーマンス最適化:

• 専用スレッドプールエグゼキューター（max_workersで設定可能）
• 高性能なAPSWベース
• WALモードと接続最適化
• 高並行性シナリオに最適

非同期フレームワークに最適:

• FastAPI, Quart, Sanic（非同期Webフレームワーク）
• aiohttp（非同期HTTP クライアント/サーバー）
• Discord.py, Telegramボット（非同期ボット）
• あらゆるasyncioベースのアプリケーション

---

License

MIT License - see LICENSE for details.