mirror of https://github.com/kortix-ai/suna.git
53 lines
1.1 KiB
Python
53 lines
1.1 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
from typing import Any, Optional
|
|
|
|
|
|
class LocalKVStore:
|
|
def __init__(self, filename: str = "kvstore.json"):
|
|
self.filename = filename
|
|
self._data = {}
|
|
self._load()
|
|
|
|
def _load(self):
|
|
if os.path.exists(self.filename):
|
|
try:
|
|
with open(self.filename, "r", encoding="utf-8") as f:
|
|
self._data = json.load(f)
|
|
except Exception:
|
|
self._data = {}
|
|
else:
|
|
self._data = {}
|
|
|
|
def _save(self):
|
|
with open(self.filename, "w", encoding="utf-8") as f:
|
|
json.dump(self._data, f, indent=2)
|
|
|
|
def get(self, key: str, default: Optional[Any] = None) -> Any:
|
|
return self._data.get(key, default)
|
|
|
|
def set(self, key: str, value: Any):
|
|
self._data[key] = value
|
|
self._save()
|
|
|
|
def delete(self, key: str):
|
|
if key in self._data:
|
|
del self._data[key]
|
|
self._save()
|
|
|
|
def clear(self):
|
|
self._data = {}
|
|
self._save()
|
|
|
|
|
|
kv = LocalKVStore()
|
|
|
|
|
|
async def main():
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|