37 lines
984 B
Python
37 lines
984 B
Python
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from typing import Generic, TypeVar
|
|
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
T = TypeVar("T", bound=BaseModel)
|
|
|
|
|
|
class DataManager(Generic[T]):
|
|
def __init__(self, cls: type[T], fp: Path) -> None:
|
|
self.cls = cls
|
|
self.fp = fp
|
|
self._aio_lock = asyncio.Lock()
|
|
self._data: T | None = None
|
|
|
|
def load(self) -> T:
|
|
if not self.fp.exists():
|
|
return self.cls()
|
|
try:
|
|
return self.cls.model_validate_json(self.fp.read_text("utf-8"))
|
|
except ValidationError:
|
|
return self.cls()
|
|
|
|
def save(self, data: T):
|
|
self.fp.write_text(data.model_dump_json(), "utf-8")
|
|
|
|
@asynccontextmanager
|
|
async def get_data(self):
|
|
await self._aio_lock.acquire()
|
|
self._data = self.load()
|
|
yield self._data
|
|
self.save(self._data)
|
|
self._data = None
|
|
self._aio_lock.release()
|