yt-notify/database.py

# -*- 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())