NyanSQLite API リファレンス
PydanticネイティブなSQLiteラッパー NyanSQLite クラスのドキュメントです。
NyanSQLite
class NyanSQLite(path: str = ':memory:', wal: bool = True, strict_deserialization: bool = False)PydanticネイティブなSQLiteラッパー。
自動的なスキーマ作成、B-treeインデックス、FTS5全文検索、
部分的な読み書き、および高度なクエリ演算子をサポートします。
バックエンドには高速な `apsw` を使用しています。
Pydantic-native SQLite wrapper.
Supports automatic schema creation, B-tree indexes, FTS5 full-text search,
Powered by the high-performance `apsw` backend.
Quick start::
class Article(BaseModel):
id: int
author: Indexed[str]
title: Searchable[str]
body: Searchable[str]
views: int = 0
db = NyanSQLite("blog.sqlite")
db.register(Article)
db.insert(Article(id=1, author="neko", title="Hello SQLite", body="…"))
db.search(Article, "SQLite")
db.update(Article, where={"id": 1}, views=42)
db.select(Article, fields=["title", "views"], author="neko")
NyanSQLiteを初期化します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
path | str | データベースファイルのパス。デフォルトはメモリ内データベース (":memory:")。 |
wal | bool | WAL (Write-Ahead Logging) モードを有効にするかどうか。デフォルトは True。 |
strict_deserialization | bool | デシリアライズ時に厳密なチェックを行うかどうか。 Trueの場合、不正なデータに対して ValueError を発生させます。 Falseの場合、警告を出して生の値を返します。 |
コンストラクタ
コアメソッド
register
def register(model: type[BaseModel]) -> NonePydanticモデルを登録し、対応するテーブル、インデックス、FTS5仮想テーブルを作成します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | 登録するPydanticモデルクラス。 |
例外
- TableNameCollisionError: 同じテーブル名を持つ別のモデルが既に登録されている場合に発生します。
close
def close() -> None基盤となるデータベース接続を閉じます。
registered_models
def registered_models() -> list[str]登録されているすべてのモデル名を取得します。
戻り値
Type: list[str]
list[str]: モデル名のリスト。
CRUD操作
insert
def insert(obj: M) -> MPydanticモデルのインスタンスをデータベースに挿入します。 Pydanticによる検証後、INSERTします。オブジェクトは変更されずに返されます。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
obj | M | 挿入するモデルのインスタンス。 |
戻り値
Type: M
M: 挿入されたオブジェクト（変更なし）。
例外
- ModelNotRegisteredError: モデルが登録されていない場合に発生します。
insert_many
def insert_many(objs: list[M]) -> int複数のモデルインスタンスを1つのトランザクションで一括挿入します。
SQLiteの変数バインド制限（デフォルト 32766）を考慮し、大きなデータセットは自動的に分割して挿入されます。 これにより、非常に大きなデータセットでのSQLITE_TOOBIGエラーを防ぎます。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
objs | list[M] | 挿入するモデルインスタンスのリスト。 |
戻り値
Type: int
int: 挿入された行数。
例外
- ModelNotRegisteredError: モデルが登録されていない場合に発生します。
update
def update(model: type[BaseModel], where: dict[str, Any], **fields: Any) -> int指定されたフィールドのみを更新する部分更新を行います。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | 更新対象のモデルクラス。 |
where | dict[str, Any] | 更新対象の行を特定する一致条件（例: {"id": 1}）。 |
fields | 更新する フィールド名=新しい値 のペア。 |
戻り値
Type: int
int: 更新された行数。
例外
- ModelNotRegisteredError: モデルが登録されていない場合に発生します。
- FieldNotFoundError: 指定されたフィールドがモデルに存在しない場合に発生します。
使用例
db.update(User, where={"id": 1}, age=26, bio="updated")delete
def delete(model: type[BaseModel], *filters: str, **kwargs: Any) -> intフィルタ条件に一致するすべての行を削除します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | 削除対象のモデルクラス。 |
filters | str | 文字列形式のフィルタ条件（例: "age > 50"）。 |
kwargs | キーワード形式のフィルタ条件（例: id=42）。 |
戻り値
Type: int
int: 削除された行数。
使用例
db.delete(User, id=42)
db.delete(User, "age > 50")
db.delete(Session, user_id=1, active=True)クエリ & 検索
get
def get(model: type[M], *filters: str, **kwargs: Any) -> Optional[M]条件に一致する最初の行をPydanticモデルとして取得します。一致しない場合は None を返します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[M] | 取得対象のモデルクラス。 |
filters | str | 文字列形式のフィルタ条件。 |
kwargs | キーワード形式のフィルタ条件。 |
戻り値
Type: Optional[M]
Optional[M]: 取得されたモデルインスタンス、または None。
使用例
user = db.get(User, id=1)
user = db.get(User, "age > 30", name="Alice")
user = db.get(User, email="[email protected]")query
def query(model: type[M], *filters: str, limit: Optional[int] = None, offset: Optional[int] = None, order_by: Optional[str] = None, desc: bool = False, **kwargs: Any) -> list[M]フィルタリング、ソート、ページネーションを使用して行を検索します。
文字列フィルタおよび演算子サフィックス（__gt, __like など）をサポートしています。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[M] | 検索対象のモデルクラス。 |
filters | str | 文字列形式のフィルタ条件。 |
limit | Optional[int] | 取得する最大行数。 |
offset | Optional[int] | 取得を開始するオフセット行数。 |
order_by | Optional[str] | ソートに使用するフィールド名。 |
desc | bool | 降順でソートするかどうか。デフォルトは False（昇順）。 **kwargs (Any): キーワード形式のフィルタ条件。 |
kwargs | キーワード形式のフィルタ条件。 |
戻り値
Type: list[M]
list[M]: 取得されたモデルインスタンスのリスト。
使用例
db.query(User) # 全ての行
db.query(User, age=25) # 完全一致
db.query(User, "age > 20", limit=10) # 文字列フィルタ
db.query(User, age__gte=20, limit=10) # 演算子サフィックス
db.query(User, order_by="name", desc=True) # ソート
db.query(User, order_by="id", limit=20, offset=40) # ページネーションselect
def select(model: type[BaseModel], fields: list[str], *filters: str, limit: Optional[int] = None, offset: Optional[int] = None, order_by: Optional[str] = None, desc: bool = False, **kwargs: Any) -> list[dict[str, Any]]特定のフィールドのみを辞書のリストとして取得します（部分読み込み）。
大きな行を持つテーブルで、未使用の列をロードするのを避けることができます。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | 取得対象のモデルクラス。 |
fields | list[str] | 取得するフィールド名のリスト。 |
filters | str | 文字列形式のフィルタ条件。 |
limit | Optional[int] | 取得する最大行数。 |
offset | Optional[int] | オフセット。 |
order_by | Optional[str] | ソートに使用するフィールド名。 |
desc | bool | 降順にするかどうか。 **kwargs (Any): キーワード形式のフィルタ条件。 |
kwargs | キーワード形式のフィルタ条件。 |
戻り値
Type: list[dict[str, Any]]
list[dict[str, Any]]: 指定されたフィールドを含む辞書のリスト。
使用例
db.select(Article, ["title", "views"], author="neko", order_by="views", desc=True)
db.select(Article, ["title"], "views > 100")search
def search(model: type[M], query: str, *, limit: Optional[int] = None) -> list[M]すべての Searchable[str] フィールドに対して全文検索を実行します。
FTS5の MATCH を使用し、BM25アルゴリズムでランク付け（ORDER BY rank）されます。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[M] | 検索対象のモデルクラス。 |
query | str | FTS5検索クエリ文字列。 limit (Optional[int]): 取得する最大行数。 |
limit | Optional[int] | 取得する最大行数。 |
戻り値
Type: list[M]
list[M]: 検索結果に一致するモデルインスタンスのリスト（関連度順）。
例外
- SearchNotEnabledError: モデルに
Searchable[str]フィールドが定義されていない場合に発生します。
使用例
db.search(Article, "python sqlite")
db.search(Article, "python sqlite", limit=5)フィールドを限定した検索には FTS5 のカラム指定構文が使用できます:
db.search(Article, "title:python")count
def count(model: type[BaseModel], *filters: str, **kwargs: Any) -> intフィルタ条件に一致する行数を返します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | カウント対象のモデルクラス。 |
filters | str | 文字列フィルタ。 |
kwargs | キーワードフィルタ。 |
戻り値
Type: int
int: 条件に一致した行数。
使用例
total = db.count(User)
adults = db.count(User, "age >= 18")
adults = db.count(User, age__gte=18)exists
def exists(model: type[BaseModel], *filters: str, **kwargs: Any) -> boolフィルタ条件に一致する行が少なくとも1つ存在するかどうかを返します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | 確認対象のモデルクラス。 |
filters | str | 文字列フィルタ。 |
kwargs | キーワードフィルタ。 |
戻り値
Type: bool
bool: 存在する場合は True、そうでない場合は False。
使用例
if db.exists(User, email="[email protected]"):
...メンテナンス
rebuild_fts
def rebuild_fts(model: type[BaseModel]) -> Noneモデルの FTS5 インデックスを再構築します（大量のデータインポート後などに有用です）。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
model | type[BaseModel] | インデックスを再構築するモデルクラス。 |
vacuum
def vacuum() -> Noneデータベースを VACUUM してディスク領域を解放します。
生のSQL実行
execute_raw
def execute_raw(sql: str, params: tuple = ()) -> list[dict[str, Any]]任意のSQLを実行し、結果を辞書のリストとして返します。
引数名
| 引数名 | 型 | 説明 |
|---|---|---|
sql | str | 実行するSQL文。 |
params | tuple | SQL文に渡すパラメータ。 |
戻り値
Type: list[dict[str, Any]]
list[dict[str, Any]]: 結果行のリスト（各行は辞書）。
使用例
db.execute_raw("SELECT count(*) AS n FROM user WHERE age > ?", (18,))その他のメソッド
atomic
def atomic() -> Generator[None, None, None]トランザクション（ATOMICブロック）を開始します。 ネスト（入れ子）された呼び出しも安全に処理されます。
使用例
with db.atomic():
db.insert(User(id=1, name="Taro"))
# ここで例外が発生するとロールバックされます