Source code for bluebase.pf.file

from typing import (
    Self,
    ClassVar,
    BinaryIO,
    TYPE_CHECKING,
)
if TYPE_CHECKING:
    from .cache import PfPageCache

from pathlib import Path

from bluebase.core import (
    LayoutField,
    LayoutMixin,
    assignment,
)

from .error import (
    PfInvalidPageIdError,
    PfFreePageAccessError,
    PfFileHeaderDisposeError,
    PfPinnedPageDisposeError,
    PfFreePageDisposeError,
)
from .type import (
    PfPageId,
    PfFileId,
)
from .page import (
    PfPageView,
    PfPageHeader,
    PfPage,
)


__all__ = (
    'PfFileHeader',
    'PfFile',
)




class PfFileHeader(PfPageView, LayoutMixin):
    """
    file header를 관리하는 view 클래스입니다.

    .. rubric:: Layout

    .. list-table::
       :header-rows: 1

       * - Offset
         - Size
         - Name
         - Format
       * - 0
         - 4
         - `page_count`
         - ``'<i'``
       * - 4
         - 4
         - `first_free_pid`
         - ``'<i'``
    """

    PID: ClassVar[PfPageId] = PfPageId.first()
    """file header의 page id를 나타내는 상수."""

    NO_FREE_PAGE: ClassVar[PfPageId] = PfPageHeader.NO_NEXT_FREE_PAGE
    """free page가 없을 경우 사용하는 상수."""

    fields = (
        LayoutField('page_count', 'i'),
        LayoutField('first_free_pid', 'i'),
    )





class PfFile:
    """
    Attributes:
        fid:
        cache:
        stream:
    """

    fid: PfFileId
    cache: 'PfPageCache'
    stream: BinaryIO

    def __repr__(self) -> str:  # pragma: no cover
        return f"PfFile({self.fid!r})"

    def __init__(self, fid: PfFileId, cache: 'PfPageCache') -> None:
        self.fid = fid
        self.stream = Path(fid).open('rb+')
        self.cache = cache



    @staticmethod
    def get_header(page: PfPage) -> PfFileHeader:
        """
        """
        return PfFileHeader(
            page,
            offset=PfPageHeader.size,
            limit=PfFileHeader.size,
        )




    @staticmethod
    def fetch_header(file: 'PfFile') -> tuple[PfPage, PfFileHeader]:
        """
        """
        page = file.cache.fetch(file, PfFileHeader.PID)
        header = PfFile.get_header(page)
        return page, header




    @classmethod
    def new(cls, fid: PfFileId, cache: 'PfPageCache') -> Self:
        """
        """
        file = cls(fid, cache)
        page = cache.acquire(file, PfFileHeader.PID)
        page.recycle()
        header = PfFile.get_header(page)
        header.set('page_count', 1)
        header.set('first_free_pid', int(PfFileHeader.NO_FREE_PAGE))
        header.commit()
        page.unpin()
        return file




    @classmethod
    def open(cls, fid: PfFileId, cache: 'PfPageCache') -> Self:
        """
        """
        file = cls(fid, cache)
        return file


    @property
    def page_count(self) -> int:
        """
        """
        page, header = PfFile.fetch_header(self)
        page_count = header.get('page_count')
        page.unpin()
        return page_count

    @page_count.setter
    def page_count(self, count: int) -> None:
        page, header = PfFile.fetch_header(self)
        header.set('page_count', count)
        header.commit()
        page.unpin()

    @property
    def first_free_pid(self) -> PfPageId | None:
        """
        """
        page, header = PfFile.fetch_header(self)
        pid = header.get('first_free_pid')
        page.unpin()
        if pid == int(PfFileHeader.NO_FREE_PAGE):
            return None
        return PfPageId(pid)

    @first_free_pid.setter
    def first_free_pid(self, pid: PfPageId | None) -> None:
        if pid is None:
            pid = PfFileHeader.NO_FREE_PAGE
        page, header = PfFile.fetch_header(self)
        header.set('first_free_pid', int(pid))
        header.commit()
        page.unpin()



    def validate_pid(self, pid: PfPageId) -> None:
        """
        Raises:
            PfInvalidPageIdError:
        """
        page_count = self.page_count
        if int(pid) < 0 or int(pid) >= page_count:
            raise PfInvalidPageIdError(file=self, pid=pid, count=page_count)




    @assignment
    def get_page(self, pid: PfPageId) -> PfPage:
        """
        Raises:
            PfInvalidPageIdError:
            PfFreePageAccessError:
        """
        raise NotImplementedError




    @assignment
    def find_next_page(self, pid: PfPageId) -> PfPage | None:
        """
        Raises:
            PfInvalidPageIdError:
            PfFreePageAccessError:
        """
        raise NotImplementedError




    @assignment
    def allocate_page(self) -> PfPage:
        """
        """
        raise NotImplementedError




    @assignment
    def dispose_page(self, page: PfPage) -> None:
        """
        Raises:
            PfFileHeaderDisposeError:
            PfPinnedPageDisposeError:
            PfFreePageDisposeError:
        """
        raise NotImplementedError




    @assignment
    def persist(self) -> None:
        """
        """
        raise NotImplementedError




    @assignment
    def flush(self) -> None:
        """
        """
        raise NotImplementedError