TinyDB bridge implementation for asyncio-based applications.
AIOBridge is an async-safe adapter for TinyDB, inspired by its design and intended for use within asyncio-based concurrent tasks. It enables safe and structured interaction with TinyDB from asynchronous Python code.
- Implements an async context manager for automatic resource handling
- Ensures concurrency safety via a shared
asyncio.Lockper DB file path - Executes all TinyDB operations using
asyncio.to_thread()for non-blocking behavior - Provides functional-style error handling via
Resultobjects - Supports configurable per-operation timeouts for robustness under load
Install via pip:
pip install tinybridgeOr using uv:
uv add tinybridge
tinybridgedepends ontinydbandresult, both installed automatically as dependencies.
AIOBridge accepts the following options during initialization:
| Parameter | Type | Default | Description |
|---|---|---|---|
path |
str |
— | Path to the TinyDB JSON file |
timeout |
int |
10 |
Timeout (in seconds) applied to each operation |
tinydb_class |
type |
TinyDB |
Optional class to override the default TinyDB implementation |
**kwargs |
dict |
— | Additional keyword arguments passed to the TinyDB constructor |
You can override the default TinyDB class either by passing a custom class directly or by subclassing AIOBridge.
from tinybridge import AIOBridge
from tinydb import TinyDB
from tinydb.storages import MemoryStorage
class InMemoryTinyDB(TinyDB):
def __init__(self, *args, **kwargs):
super().__init__(*args, storage=MemoryStorage, **kwargs)
async with AIOBridge("db.json", tinydb_class=InMemoryTinyDB) as db:
...from tinybridge import AIOBridge
from tinydb import TinyDB
from tinydb.storages import JSONStorage
from tinydb.middlewares import CachingMiddleware
class CustomTinyDB(TinyDB):
def __init__(self, *args, **kwargs):
super().__init__(*args, storage=CachingMiddleware(JSONStorage), **kwargs)
class CustomAIOBridge(AIOBridge):
tinydb_class = CustomTinyDBMinimal example demonstrating asynchronous insert:
import asyncio
from tinybridge import AIOBridge
async def main():
async with AIOBridge("db.json") as db:
result = await db.insert({"name": "Alice"})
if result.is_ok():
print("Inserted:", result.unwrap())
else:
print("Insert failed:", result.unwrap_err())
asyncio.run(main())All standard TinyDB operations—such as get, search, update, and remove—are supported. Each method is wrapped for async compatibility and returns a Result object for safe error handling.
Alternative community-driven efforts to add async capabilities to TinyDB: