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) -> NoneAsync 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() -> NoneExample
await db.close()table
def table(table_name: str, validator: Any | None | EllipsisType = Ellipsis, coerce: bool | EllipsisType = Ellipsis, hooks: list[NanaHook] | None | EllipsisType = Ellipsis) -> AsyncNanaSQLitesub1 = 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 |
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) -> AnyParameter
| 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() -> dictReturns
Type: dict
Example
data = await db.to_dict()copy
def copy() -> dictReturns
Type: dict
Example
data_copy = await db.copy()clear_cache
def clear_cache() -> NoneData Management
aflush
def aflush(wait: bool = False) -> NoneExample
await db.aflush(wait=True)flush
def flush(wait: bool = False) -> NoneExample
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() -> NoneExample
await db.aretry_dlq()retry_dlq
def retry_dlq() -> NoneExample
await db.aretry_dlq()aclear_dlq
def aclear_dlq() -> NoneExample
await db.aclear_dlq()clear_dlq
def clear_dlq() -> NoneExample
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() -> NoneExample
await db.load_all()refresh
def refresh(key: str | None = None) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
key | `str | None` |
Example
await db.refresh("user")
await db.refresh() # 全キャッシュ更新is_cached
def is_cached(key: str) -> boolParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Type: bool
Example
cached = await db.is_cached("user")batch_update
def batch_update(mapping: dict[str, Any]) -> NoneParameter
| 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] |
batch_delete
def batch_delete(keys: list[str]) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
keys | list[str] |
Example
await db.batch_delete(["key1", "key2", "key3"])get_fresh
def get_fresh(key: str, default: Any = None) -> AnyParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Example
value = await db.get_fresh("key")Transaction Control
begin_transaction
def begin_transaction() -> NoneExample
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() -> NoneExample
await db.commit()rollback
def rollback() -> NoneExample
await db.rollback()in_transaction
def in_transaction() -> boolReturns
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) -> NoneParameter
| 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) -> intParameter
| 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) -> intParameter
| 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) -> intParameter
| 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) -> intParameter
| 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) -> AnyParameter
| 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]) -> NoneParameter
| 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 | NoneParameter
| 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) -> NoneParameter
| 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) -> NoneParameter
| 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) -> boolParameter
| 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) -> NoneParameter
| 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) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
index_name | str | |
if_exists | bool |
Example
await db.drop_index("idx_users_email")Utility Functions
vacuum
def vacuum() -> NoneExample
await db.vacuum()Pydantic Support
set_model
def set_model(key: str, model: Any) -> NoneParameter
| 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) -> AnyParameter
| 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
def aget(key: str, default: Any = None) -> AnyParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Example
user = await db.aget("user")
config = await db.aget("config", {})aset
def aset(key: str, value: Any) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Example
await db.aset("user", {"name": "Nana", "age": 20})adelete
def adelete(key: str) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Raises
Example
await db.adelete("old_data")acontains
def acontains(key: str) -> boolParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Type: bool
Example
if await db.acontains("user"):
print("User exists")contains
def contains(key: str) -> boolParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Type: bool
Example
if await db.acontains("user"):
print("User exists")alen
def alen() -> intReturns
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) -> AnyParameter
| 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) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
mapping | `dict | None` |
Example
await db.aupdate({"key1": "value1", "key2": "value2"})
await db.aupdate(key3="value3", key4="value4")aclear
def aclear() -> NoneExample
await db.aclear()asetdefault
def asetdefault(key: str, default: Any = None) -> AnyParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Example
value = await db.asetdefault("config", {})aload_all
def aload_all() -> NoneExample
await db.load_all()arefresh
def arefresh(key: str | None = None) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
key | `str | None` |
Example
await db.refresh("user")
await db.refresh() # 全キャッシュ更新ais_cached
def ais_cached(key: str) -> boolParameter
| Parameter | Type | Description |
|---|---|---|
key | str |
Returns
Type: bool
Example
cached = await db.is_cached("user")abatch_update
def abatch_update(mapping: dict[str, Any]) -> NoneParameter
| 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]) -> NoneParameter
| Parameter | Type | Description |
|---|---|---|
keys | list[str] |
Example
await db.batch_delete(["key1", "key2", "key3"])ato_dict
def ato_dict() -> dictReturns
Type: dict
Example
data = await db.to_dict()acopy
def acopy() -> dictReturns
Type: dict
Example
data_copy = await db.copy()aget_fresh
def aget_fresh(key: str, default: Any = None) -> AnyParameter
| 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) -> NoneParameter
| 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) -> AnyParameter
| 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) -> AnyParameter
| 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]) -> NoneParameter
| 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 | NoneParameter
| 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) -> NoneParameter
| 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) -> NoneParameter
| 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) -> boolParameter
| 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) -> NoneParameter
| 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) -> intParameter
| 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) -> intParameter
| 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) -> intParameter
| 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) -> intParameter
| 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() -> NoneExample
await db.vacuum()aclear_cache
def aclear_cache() -> Noneatable
def atable(table_name: str, validator: Any | None | EllipsisType = Ellipsis, coerce: bool | EllipsisType = Ellipsis, hooks: list[NanaHook] | None | EllipsisType = Ellipsis) -> AsyncNanaSQLitesub1 = 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 |
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) -> Nonearestore
def arestore(source_path: str) -> Noneapragma
def apragma(pragma_name: str, value: Any = None) -> AnyParameter
| 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) -> Noneaupsert
def aupsert(table_name: str | Any = None, data: Any = None, conflict_columns: list[str] | None = None) -> int | NoneParameter
| Parameter | Type | Description |
|---|---|---|
table_name | `str | Any` |
conflict_columns | `list[str] | None` |