Asynchronous API Reference

Reference for the asynchronous AsyncNanaSQLite class.

AsyncNanaSQLite

class AsyncNanaSQLite(db_path: str, table: str = 'data', bulk_load: bool = False, optimize: bool = True, cache_size_mb: int = 64, max_workers: int = 5, strict_sql_validation: bool = True, validator: Any | None = None, coerce: bool = False, v2_mode: bool = False, v2_config: V2Config | None = None, **kwargs: Any) -> None

Async wrapper for NanaSQLite with optimized thread pool executor.

All database operations are executed in a dedicated thread pool executor to prevent blocking the async event loop. This allows NanaSQLite to be used safely in async applications like FastAPI, aiohttp, etc.

The implementation uses a configurable thread pool for optimal concurrency and performance in high-load scenarios.

Parameter

Parameter	Type	Description
db_path	str
table	str
bulk_load	bool
optimize	bool
cache_size_mb	int
max_workers	int
strict_sql_validation	bool
validator	`Any	None`
coerce	bool
v2_mode	bool
v2_config	`V2Config	None`

Example

    async with AsyncNanaSQLite("mydata.db") as db:
        await db.aset("config", {"theme": "dark"})
        config = await db.aget("config")
        print(config)

    # 高負荷環境向けの設定
    async with AsyncNanaSQLite("mydata.db", max_workers=10) as db:
        # 並行処理が多い場合に最適化
        results = await asyncio.gather(*[db.aget(f"key_{i}") for i in range(100)])

---

Constructor

Core Methods

close

def close() -> None

Drain accepted work and close; caller cancellation is shielded.

---

table

def table(table_name: str, validator: Any | None | EllipsisType = Ellipsis, coerce: bool | EllipsisType = Ellipsis, hooks: list[NanaHook] | None | EllipsisType = Ellipsis, warn_duplicate_table_instance: bool | EllipsisType = Ellipsis) -> AsyncNanaSQLite

sub1 = await db.table("users")

users_db = await db.table("users")

Parameter

Parameter	Type	Description
table_name	str
validator	`Any	None
coerce	`bool	EllipsisType`
hooks	`list[NanaHook]	None
warn_duplicate_table_instance	`bool	EllipsisType`

Returns

Type: AsyncNanaSQLite

Example

    async with AsyncNanaSQLite("mydata.db", table="main") as db:
        users_db = await db.table("users")
        products_db = await db.table("products")
        await users_db.aset("user1", {"name": "Alice"})
        await products_db.aset("prod1", {"name": "Laptop"})

---

Dictionary Interface

get

def get(key: str, default: Any = None) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    user = await db.aget("user")
    config = await db.aget("config", {})

---

keys

def keys() -> list[str]

Returns

Type: list[str]

Example

    keys = await db.akeys()

---

values

def values() -> list[Any]

Returns

Type: list[Any]

Example

    values = await db.avalues()

---

items

def items() -> list[tuple[str, Any]]

Returns

Type: list[tuple[str, Any]]

Example

    items = await db.aitems()

---

to_dict

def to_dict() -> dict

Returns

Type: dict

Example

    data = await db.to_dict()

---

copy

def copy() -> dict

Returns

Type: dict

Example

    data_copy = await db.copy()

---

clear_cache

def clear_cache() -> None

---

Data Management

aflush

def aflush(wait: bool = False) -> None

Example

    await db.aflush(wait=True)

---

flush

def flush(wait: bool = False) -> None

Example

    await db.aflush(wait=True)

---

aget_dlq

def aget_dlq() -> list[dict[str, Any]]

Example

    failed = await db.aget_dlq()

---

get_dlq

def get_dlq() -> list[dict[str, Any]]

Example

    failed = await db.aget_dlq()

---

aretry_dlq

def aretry_dlq() -> None

Example

    await db.aretry_dlq()

---

retry_dlq

def retry_dlq() -> None

Example

    await db.aretry_dlq()

---

aclear_dlq

def aclear_dlq() -> None

Example

    await db.aclear_dlq()

---

clear_dlq

def clear_dlq() -> None

Example

    await db.aclear_dlq()

---

aget_v2_metrics

def aget_v2_metrics() -> dict[str, Any]

Example

    metrics = await db.aget_v2_metrics()
    print(metrics["flush_count"])

---

get_v2_metrics

def get_v2_metrics() -> dict[str, Any]

Example

    metrics = await db.aget_v2_metrics()
    print(metrics["flush_count"])

---

load_all

def load_all() -> None

Example

    await db.load_all()

---

refresh

def refresh(key: str | None = None) -> None

Parameter

Parameter	Type	Description
key	`str	None`

Example

    await db.refresh("user")
    await db.refresh()  # 全キャッシュ更新

---

is_cached

def is_cached(key: str) -> bool

Parameter

Parameter	Type	Description
key	str

Returns

Type: bool

Example

    cached = await db.is_cached("user")

---

batch_update

def batch_update(mapping: dict[str, Any]) -> None

Parameter

Parameter	Type	Description
mapping	dict[str, Any]

Example

    await db.batch_update({
        "key1": "value1",
        "key2": "value2",
        "key3": {"nested": "data"}
    })

---

batch_update_partial

def batch_update_partial(mapping: dict[str, Any]) -> dict[str, str]

Parameter

Parameter	Type	Description
mapping	dict[str, Any]

---

aincrement

def aincrement(key: str, amount: int | float = 1, *, field: str | None = None, default: int | float | None = None) -> Any

Atomically increment a numeric value or top-level numeric dict field.

---

increment

def increment(key: str, amount: int | float = 1, *, field: str | None = None, default: int | float | None = None) -> Any

Atomically increment a numeric value or top-level numeric dict field.

---

apatch

def apatch(key: str, changes: dict[str, Any], *, create: bool = False) -> Any

Atomically apply a shallow dictionary merge and return the stored value.

---

patch

def patch(key: str, changes: dict[str, Any], *, create: bool = False) -> Any

Atomically apply a shallow dictionary merge and return the stored value.

---

batch_delete

def batch_delete(keys: list[str]) -> None

Parameter

Parameter	Type	Description
keys	list[str]

Example

    await db.batch_delete(["key1", "key2", "key3"])

---

get_fresh

def get_fresh(key: str, default: Any = None) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    value = await db.get_fresh("key")

---

Transaction Control

begin_transaction

def begin_transaction() -> None

Example

    await db.begin_transaction()
    try:
        await db.sql_insert("users", {"name": "Alice"})
        await db.sql_insert("users", {"name": "Bob"})
        await db.commit()
    except:
        await db.rollback()

---

commit

def commit() -> None

Example

    await db.commit()

---

rollback

def rollback() -> None

Example

    await db.rollback()

---

in_transaction

def in_transaction() -> bool

Returns

Type: bool

Example

    status = await db.in_transaction()
    print(f"In transaction: {status}")

---

transaction

def transaction()

Example

    async with db.transaction():
        await db.sql_insert("users", {"name": "Alice"})
        await db.sql_insert("users", {"name": "Bob"})
        # 自動的にコミット、例外時はロールバック

---

SQL Wrapper (CRUD)

upsert

def upsert(table_name: str | None = None, data: dict[str, Any] | Any | None = None, *, conflict_columns: list[str] | None = None) -> None

Parameter

Parameter	Type	Description
table_name	`str	None`
data	`dict[str, Any]	Any
conflict_columns	`list[str]	None`

Example

    await db.aupsert("users", {"id": 1, "name": "Alice"})
    await db.aupsert("user:1", {"name": "Nana"})

---

sql_insert

def sql_insert(table_name: str, data: dict) -> int

Parameter

Parameter	Type	Description
table_name	str
data	dict

Returns

Type: int

Example

    rowid = await db.sql_insert("users", {
        "name": "Alice",
        "email": "[email protected]",
        "age": 25
    })

---

sql_update

def sql_update(table_name: str, data: dict, where: str, parameters: tuple | None = None) -> int

Parameter

Parameter	Type	Description
table_name	str
data	dict
where	str
parameters	`tuple	None`

Returns

Type: int

Example

    count = await db.sql_update("users",
        {"age": 26, "status": "active"},
        "name = ?",
        ("Alice",)
    )

---

sql_delete

def sql_delete(table_name: str, where: str, parameters: tuple | None = None) -> int

Parameter

Parameter	Type	Description
table_name	str
where	str
parameters	`tuple	None`

Returns

Type: int

Example

    count = await db.sql_delete("users", "age < ?", (18,))

---

Query

query

def query(table_name: str | None = None, columns: list[str] | None = None, where: str | None = None, parameters: tuple | None = None, order_by: str | None = None, limit: int | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> list[dict]

Parameter

Parameter	Type	Description
table_name	`str	None`
columns	`list[str]	None`
where	`str	None`
parameters	`tuple	None`
order_by	`str	None`
limit	`int	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: list[dict]

Example

    results = await db.query(
        table_name="users",
        columns=["id", "name", "email"],
        where="age > ?",
        parameters=(20,),
        order_by="name ASC",
        limit=10
    )

---

query_with_pagination

def query_with_pagination(table_name: str | None = None, columns: list[str] | None = None, where: str | None = None, parameters: tuple | None = None, order_by: str | None = None, limit: int | None = None, offset: int | None = None, group_by: str | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> list[dict]

Parameter

Parameter	Type	Description
table_name	`str	None`
columns	`list[str]	None`
where	`str	None`
parameters	`tuple	None`
order_by	`str	None`
limit	`int	None`
offset	`int	None`
group_by	`str	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: list[dict]

Example

    results = await db.query_with_pagination(
        table_name="users",
        columns=["id", "name", "email"],
        where="age > ?",
        parameters=(20,),
        order_by="name ASC",
        limit=10,
        offset=0
    )

---

count

def count(table_name: str | None = None, where: str | None = None, parameters: tuple | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> int

Parameter

Parameter	Type	Description
table_name	`str	None`
where	`str	None`
parameters	`tuple	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: int

Example

    count = await db.count("users", "age < ?", (18,))

---

Direct SQL Execution

execute

def execute(sql: str, parameters: tuple | None = None) -> Any

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Example

    cursor = await db.execute("SELECT * FROM data WHERE key LIKE ?", ("user%",))

---

execute_many

def execute_many(sql: str, parameters_list: list[tuple]) -> None

Parameter

Parameter	Type	Description
sql	str
parameters_list	list[tuple]

Example

    await db.execute_many(
        "INSERT OR REPLACE INTO custom (id, name) VALUES (?, ?)",
        [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
    )

---

fetch_one

def fetch_one(sql: str, parameters: tuple | None = None) -> tuple | None

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Type: tuple | None

Example

    row = await db.fetch_one("SELECT value FROM data WHERE key = ?", ("user",))

---

fetch_all

def fetch_all(sql: str, parameters: tuple | None = None) -> list[tuple]

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Type: list[tuple]

Example

    rows = await db.fetch_all("SELECT key, value FROM data WHERE key LIKE ?", ("user%",))

---

Schema Management

create_table

def create_table(table_name: str, columns: dict, if_not_exists: bool = True, primary_key: str | None = None) -> None

Parameter

Parameter	Type	Description
table_name	str
columns	dict
if_not_exists	bool
primary_key	`str	None`

Example

    await db.create_table("users", {
        "id": "INTEGER PRIMARY KEY",
        "name": "TEXT NOT NULL",
        "email": "TEXT UNIQUE"
    })

---

create_index

def create_index(index_name: str, table_name: str, columns: list[str], unique: bool = False, if_not_exists: bool = True) -> None

Parameter

Parameter	Type	Description
index_name	str
table_name	str
columns	list[str]
unique	bool
if_not_exists	bool

Example

    await db.create_index("idx_users_email", "users", ["email"], unique=True)

---

table_exists

def table_exists(table_name: str) -> bool

Parameter

Parameter	Type	Description
table_name	str

Returns

Type: bool

Example

    exists = await db.table_exists("users")

---

list_tables

def list_tables() -> list[str]

Returns

Type: list[str]

Example

    tables = await db.list_tables()

---

drop_table

def drop_table(table_name: str, if_exists: bool = True) -> None

Parameter

Parameter	Type	Description
table_name	str
if_exists	bool

Example

    await db.drop_table("old_table")

---

drop_index

def drop_index(index_name: str, if_exists: bool = True) -> None

Parameter

Parameter	Type	Description
index_name	str
if_exists	bool

Example

    await db.drop_index("idx_users_email")

---

Utility Functions

vacuum

def vacuum() -> None

Example

    await db.vacuum()

---

Pydantic Support

set_model

def set_model(key: str, model: Any) -> None

Parameter

Parameter	Type	Description
key	str

Example

    from pydantic import BaseModel
    class User(BaseModel):
        name: str
        age: int
    user = User(name="Nana", age=20)
    await db.set_model("user", user)

---

get_model

def get_model(key: str, model_class: type = None) -> Any

Parameter

Parameter	Type	Description
key	str
model_class	type

Returns

Example

    user = await db.get_model("user", User)

---

Other Methods

add_hook

def add_hook(hook: NanaHook) -> None

[v1.5.0 Feature] Add a hook/constraint to intercept read, write, and delete operations.

Parameter

Parameter	Type	Description
hook	NanaHook	Any object implementing the NanaHook protocol.

---

aget_status

def aget_status() -> dict[str, Any]

Return payload-free persistence diagnostics.

---

aget_dlq_summary

def aget_dlq_summary() -> list[dict[str, Any]]

Return redacted failure codes and timestamps.

---

aiter_items

def aiter_items(batch_size: int = 1000) -> AsyncIterator[tuple[str, Any]]

Async keyset iteration; see NanaSQLite.iter_items for consistency rules.

---

aget

def aget(key: str, default: Any = None) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    user = await db.aget("user")
    config = await db.aget("config", {})

---

aset

def aset(key: str, value: Any) -> None

Parameter

Parameter	Type	Description
key	str

Example

    await db.aset("user", {"name": "Nana", "age": 20})

---

adelete

def adelete(key: str) -> None

Parameter

Parameter	Type	Description
key	str

Raises

Example

    await db.adelete("old_data")

---

acontains

def acontains(key: str) -> bool

Parameter

Parameter	Type	Description
key	str

Returns

Type: bool

Example

    if await db.acontains("user"):
        print("User exists")

---

contains

def contains(key: str) -> bool

Parameter

Parameter	Type	Description
key	str

Returns

Type: bool

Example

    if await db.acontains("user"):
        print("User exists")

---

alen

def alen() -> int

Returns

Type: int

Example

    count = await db.alen()

---

akeys

def akeys() -> list[str]

Returns

Type: list[str]

Example

    keys = await db.akeys()

---

avalues

def avalues() -> list[Any]

Returns

Type: list[Any]

Example

    values = await db.avalues()

---

aitems

def aitems() -> list[tuple[str, Any]]

Returns

Type: list[tuple[str, Any]]

Example

    items = await db.aitems()

---

apop

def apop(key: str, *args) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    value = await db.apop("temp_data")
    value = await db.apop("maybe_missing", "default")

---

aupdate

def aupdate(mapping: dict | None = None, **kwargs) -> None

Parameter

Parameter	Type	Description
mapping	`dict	None`

Example

    await db.aupdate({"key1": "value1", "key2": "value2"})
    await db.aupdate(key3="value3", key4="value4")

---

aclear

def aclear() -> None

Example

    await db.aclear()

---

asetdefault

def asetdefault(key: str, default: Any = None) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    value = await db.asetdefault("config", {})

---

aload_all

def aload_all() -> None

Example

    await db.load_all()

---

arefresh

def arefresh(key: str | None = None) -> None

Parameter

Parameter	Type	Description
key	`str	None`

Example

    await db.refresh("user")
    await db.refresh()  # 全キャッシュ更新

---

ais_cached

def ais_cached(key: str) -> bool

Parameter

Parameter	Type	Description
key	str

Returns

Type: bool

Example

    cached = await db.is_cached("user")

---

abatch_update

def abatch_update(mapping: dict[str, Any]) -> None

Parameter

Parameter	Type	Description
mapping	dict[str, Any]

Example

    await db.batch_update({
        "key1": "value1",
        "key2": "value2",
        "key3": {"nested": "data"}
    })

---

abatch_update_partial

def abatch_update_partial(mapping: dict[str, Any]) -> dict[str, str]

Parameter

Parameter	Type	Description
mapping	dict[str, Any]

---

abatch_delete

def abatch_delete(keys: list[str]) -> None

Parameter

Parameter	Type	Description
keys	list[str]

Example

    await db.batch_delete(["key1", "key2", "key3"])

---

ato_dict

def ato_dict() -> dict

Returns

Type: dict

Example

    data = await db.to_dict()

---

acopy

def acopy() -> dict

Returns

Type: dict

Example

    data_copy = await db.copy()

---

aget_fresh

def aget_fresh(key: str, default: Any = None) -> Any

Parameter

Parameter	Type	Description
key	str

Returns

Example

    value = await db.get_fresh("key")

---

abatch_get

def abatch_get(keys: list[str]) -> dict[str, Any]

Parameter

Parameter	Type	Description
keys	list[str]

Returns

Type: dict[str, Any]

Example

    results = await db.abatch_get(["key1", "key2"])

---

aset_model

def aset_model(key: str, model: Any) -> None

Parameter

Parameter	Type	Description
key	str

Example

    from pydantic import BaseModel
    class User(BaseModel):
        name: str
        age: int
    user = User(name="Nana", age=20)
    await db.set_model("user", user)

---

aget_model

def aget_model(key: str, model_class: type = None) -> Any

Parameter

Parameter	Type	Description
key	str
model_class	type

Returns

Example

    user = await db.get_model("user", User)

---

aexecute

def aexecute(sql: str, parameters: tuple | None = None) -> Any

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Example

    cursor = await db.execute("SELECT * FROM data WHERE key LIKE ?", ("user%",))

---

aexecute_many

def aexecute_many(sql: str, parameters_list: list[tuple]) -> None

Parameter

Parameter	Type	Description
sql	str
parameters_list	list[tuple]

Example

    await db.execute_many(
        "INSERT OR REPLACE INTO custom (id, name) VALUES (?, ?)",
        [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
    )

---

afetch_one

def afetch_one(sql: str, parameters: tuple | None = None) -> tuple | None

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Type: tuple | None

Example

    row = await db.fetch_one("SELECT value FROM data WHERE key = ?", ("user",))

---

afetch_all

def afetch_all(sql: str, parameters: tuple | None = None) -> list[tuple]

Parameter

Parameter	Type	Description
sql	str
parameters	`tuple	None`

Returns

Type: list[tuple]

Example

    rows = await db.fetch_all("SELECT key, value FROM data WHERE key LIKE ?", ("user%",))

---

acreate_table

def acreate_table(table_name: str, columns: dict, if_not_exists: bool = True, primary_key: str | None = None) -> None

Parameter

Parameter	Type	Description
table_name	str
columns	dict
if_not_exists	bool
primary_key	`str	None`

Example

    await db.create_table("users", {
        "id": "INTEGER PRIMARY KEY",
        "name": "TEXT NOT NULL",
        "email": "TEXT UNIQUE"
    })

---

acreate_index

def acreate_index(index_name: str, table_name: str, columns: list[str], unique: bool = False, if_not_exists: bool = True) -> None

Parameter

Parameter	Type	Description
index_name	str
table_name	str
columns	list[str]
unique	bool
if_not_exists	bool

Example

    await db.create_index("idx_users_email", "users", ["email"], unique=True)

---

aquery

def aquery(table_name: str | None = None, columns: list[str] | None = None, where: str | None = None, parameters: tuple | None = None, order_by: str | None = None, limit: int | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> list[dict]

Parameter

Parameter	Type	Description
table_name	`str	None`
columns	`list[str]	None`
where	`str	None`
parameters	`tuple	None`
order_by	`str	None`
limit	`int	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: list[dict]

Example

    results = await db.query(
        table_name="users",
        columns=["id", "name", "email"],
        where="age > ?",
        parameters=(20,),
        order_by="name ASC",
        limit=10
    )

---

aquery_with_pagination

def aquery_with_pagination(table_name: str | None = None, columns: list[str] | None = None, where: str | None = None, parameters: tuple | None = None, order_by: str | None = None, limit: int | None = None, offset: int | None = None, group_by: str | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> list[dict]

Parameter

Parameter	Type	Description
table_name	`str	None`
columns	`list[str]	None`
where	`str	None`
parameters	`tuple	None`
order_by	`str	None`
limit	`int	None`
offset	`int	None`
group_by	`str	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: list[dict]

Example

    results = await db.query_with_pagination(
        table_name="users",
        columns=["id", "name", "email"],
        where="age > ?",
        parameters=(20,),
        order_by="name ASC",
        limit=10,
        offset=0
    )

---

atable_exists

def atable_exists(table_name: str) -> bool

Parameter

Parameter	Type	Description
table_name	str

Returns

Type: bool

Example

    exists = await db.table_exists("users")

---

alist_tables

def alist_tables() -> list[str]

Returns

Type: list[str]

Example

    tables = await db.list_tables()

---

adrop_table

def adrop_table(table_name: str, if_exists: bool = True) -> None

Parameter

Parameter	Type	Description
table_name	str
if_exists	bool

Example

    await db.drop_table("old_table")

---

asql_insert

def asql_insert(table_name: str, data: dict) -> int

Parameter

Parameter	Type	Description
table_name	str
data	dict

Returns

Type: int

Example

    rowid = await db.sql_insert("users", {
        "name": "Alice",
        "email": "[email protected]",
        "age": 25
    })

---

asql_update

def asql_update(table_name: str, data: dict, where: str, parameters: tuple | None = None) -> int

Parameter

Parameter	Type	Description
table_name	str
data	dict
where	str
parameters	`tuple	None`

Returns

Type: int

Example

    count = await db.sql_update("users",
        {"age": 26, "status": "active"},
        "name = ?",
        ("Alice",)
    )

---

asql_delete

def asql_delete(table_name: str, where: str, parameters: tuple | None = None) -> int

Parameter

Parameter	Type	Description
table_name	str
where	str
parameters	`tuple	None`

Returns

Type: int

Example

    count = await db.sql_delete("users", "age < ?", (18,))

---

acount

def acount(table_name: str | None = None, where: str | None = None, parameters: tuple | None = None, strict_sql_validation: bool | None = None, allowed_sql_functions: list[str] | None = None, forbidden_sql_functions: list[str] | None = None, override_allowed: bool = False) -> int

Parameter

Parameter	Type	Description
table_name	`str	None`
where	`str	None`
parameters	`tuple	None`
strict_sql_validation	`bool	None`
allowed_sql_functions	`list[str]	None`
forbidden_sql_functions	`list[str]	None`
override_allowed	bool

Returns

Type: int

Example

    count = await db.count("users", "age < ?", (18,))

---

avacuum

def avacuum() -> None

Example

    await db.vacuum()

---

aclear_cache

def aclear_cache() -> None

---

atable

def atable(table_name: str, validator: Any | None | EllipsisType = Ellipsis, coerce: bool | EllipsisType = Ellipsis, hooks: list[NanaHook] | None | EllipsisType = Ellipsis, warn_duplicate_table_instance: bool | EllipsisType = Ellipsis) -> AsyncNanaSQLite

sub1 = await db.table("users")

users_db = await db.table("users")

Parameter

Parameter	Type	Description
table_name	str
validator	`Any	None
coerce	`bool	EllipsisType`
hooks	`list[NanaHook]	None
warn_duplicate_table_instance	`bool	EllipsisType`

Returns

Type: AsyncNanaSQLite

Example

    async with AsyncNanaSQLite("mydata.db", table="main") as db:
        users_db = await db.table("users")
        products_db = await db.table("products")
        await users_db.aset("user1", {"name": "Alice"})
        await products_db.aset("prod1", {"name": "Laptop"})

---

abackup

def abackup(target_path: str, *, verify: bool = True, flush: bool = True, allow_incomplete: bool = False) -> None

---

arestore

def arestore(source_path: str) -> None

---

apragma

def apragma(pragma_name: str, value: Any = None) -> Any

Parameter

Parameter	Type	Description
pragma_name	str

---

aget_table_schema

def aget_table_schema(table_name: str | None = None) -> list[dict]

Parameter

Parameter	Type	Description
table_name	`str	None`

---

alist_indexes

def alist_indexes(table_name: str | None = None) -> list[str]

Parameter

Parameter	Type	Description
table_name	`str	None`

---

aalter_table_add_column

def aalter_table_add_column(table_name: str, column_name: str, column_type: str) -> None

---

aupsert

def aupsert(table_name: str | Any = None, data: Any = None, conflict_columns: list[str] | None = None) -> int | None

Parameter

Parameter	Type	Description
table_name	`str	Any`
conflict_columns	`list[str]	None`

---