123456789101112131415161718192021222324252627282930313233343536373839404142 |
- # -*- coding: utf-8 -*-
- from pathlib import Path
- from typing import Optional
- __all__ = ['Database']
- class AbstractDatabase:
- def __init__(self, root: Path):
- self._root = Path(root).expanduser().absolute()
- def check(self, mode: int = 0o700) -> None:
- return self._root.mkdir(mode, parents=True, exist_ok=True)
- async def insert(self, key: str, value: bytes) -> int:
- with (self._root / key).open(mode='wb') as item:
- return item.write(value)
- async def exist(self, key: str) -> bool:
- return (self._root / key).exists()
- async def get(self, key: str) -> bytes:
- with (self._root / key).open(mode='rb') as item:
- return item.read()
- class Database(AbstractDatabase):
- def __init__(self, root: Path):
- super().__init__(root)
- self.check()
- async def get_last_live(self, channel: str) -> Optional[str]:
- if await self.exist(channel):
- if last := await self.get(channel):
- return last.decode()
- return None
- async def update_last_live(self, channel: str, item: str) -> int:
- return await self.insert(channel, item.encode())
|