Use S3 Storage for TinyDB#
TinyDb support to extend the storage layer. In this example, we use a JSON file on S3 as the backend.
Reference:
[8]:
import json
import tinydb
from s3pathlib import S3Path, context
from boto_session_manager import BotoSesManager
from rich import print as rprint
class S3Storage(tinydb.Storage):
def __init__(
self,
s3path: S3Path,
bsm: BotoSesManager,
):
self.s3path = s3path
self.bsm = bsm
def read(self):
try:
return json.loads(self.s3path.read_text(bsm=bsm))
except Exception as e:
if "does not exist" in str(e):
self.write({})
else:
raise e
def write(self, data):
self.s3path.write_text(
json.dumps(data, indent=4),
content_type="application/json",
bsm=self.bsm,
)
[9]:
bsm = BotoSesManager(profile_name="awshsh_app_dev_us_east_1")
context.attach_boto_session(bsm.boto_ses)
bucket = f"{bsm.aws_account_id}-{bsm.aws_region}-data"
s3path = S3Path(f"s3://{bucket}/projects/tinydb/db.json")
print(f"use {s3path.uri} as tinydb storage")
print(f"preview at: {s3path.console_url}")
_ = s3path.delete()
use s3://807388292768-us-east-1-data/projects/tinydb/db.json as tinydb storage
preview at: https://console.aws.amazon.com/s3/object/807388292768-us-east-1-data?prefix=projects/tinydb/db.json
[10]:
with tinydb.TinyDB(
s3path=s3path,
bsm=bsm,
storage=S3Storage,
) as db:
t_user = db.table("users")
t_user.insert({"name": "Alice", "age": 30})
t_user.insert({"name": "Bob", "age": 12})
t_user.insert({"name": "Cathy", "age": 56})
User = tinydb.Query()
print("--- Query | User.name == Alice ---")
rprint(t_user.search(User.name == "Alice"))
print("--- Query | User.age >= 18 ---")
rprint(t_user.search(User.age >= 18))
--- Query | User.name == Alice ---
[{'name': 'Alice', 'age': 30}]
--- Query | User.age >= 18 ---
[{'name': 'Alice', 'age': 30}, {'name': 'Cathy', 'age': 56}]
[ ]: