Source code for bluebase.rm.file

from typing import (
    Self,
    ClassVar,
)

from bluebase.core import (
    LayoutField,
    LayoutMixin,
    dump,
    assignment,
)
from bluebase.pf import (
    PfPageId,
    PfPageView,
    PfPageHeader,
    PfPage,
    PfFileHeader,
    PfFile,
)

from .error import (
    RmError,
    RmFileHeaderAccessError,
)
from .type import (
    RmPageId,
    RmRecordId,
    RmRecordData,
    RmRecord,
)
from .page import (
    RmPageHeader,
    RmPage,
)


__all__ = (
    'RmFileHeader',
    'RmFile',
)




class RmFileHeader(PfPageView, LayoutMixin):
    """
    file header를 나타내는 view 클래스입니다.

    .. rubric:: Layout

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

       * - Offset
         - Size
         - Name
         - Format
       * - 0
         - 4
         - `record_size`
         - ``'<i'``
       * - 4
         - 4
         - `record_capacity`
         - ``'<i'``
       * - 8
         - 4
         - `first_available_pid`
         - ``'<i'``
    """

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

    NO_AVAILABLE_PAGE: ClassVar[RmPageId] = RmPageHeader.NO_NEXT_AVAILABLE_PAGE
    """available page가 없을 경우 사용하는 상수."""

    fields = (
        LayoutField('record_size', 'i'),
        LayoutField('record_capacity', 'i'),
        LayoutField('first_available_pid', 'i'),
    )





class RmFile:
    """
    Attributes:
        file_:
        record_size:
        record_capacity:
    """

    file_: PfFile
    record_size: int
    record_capacity: int

    def __repr__(self) -> str:  # pragma: no cover
        return f"RmFile({dump(
            file=self.file_,
            record_size=self.record_size,
            record_capacity=self.record_capacity,
        )})"

    def __init__(self,
                 file_: PfFile,
                 record_size: int,
                 record_capacity: int,
                 ) -> None:
        self.file_ = file_
        self.record_size = record_size
        self.record_capacity = record_capacity



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




    @staticmethod
    def fetch_header(file_: PfFile) -> tuple[PfPage, RmFileHeader]:
        """
        """
        page_ = file_.get_page(RmFileHeader.PID)
        header = RmFile.get_header(page_)
        return page_, header




    @classmethod
    def new(cls, file_: PfFile, record_size: int, record_capacity: int) -> Self:
        """
        """
        page_ = file_.allocate_page()
        header = RmFile.get_header(page_)
        header.set('record_size', record_size)
        header.set('record_capacity', record_capacity)
        header.set('first_available_pid', int(RmFileHeader.NO_AVAILABLE_PAGE))
        header.commit()
        page_.unpin()
        return cls(file_, record_size, record_capacity)




    @classmethod
    def open(cls, file_: PfFile) -> Self:
        """
        """
        page_, header = RmFile.fetch_header(file_)
        record_size = header.get('record_size')
        record_capacity = header.get('record_capacity')
        page_.unpin()
        return cls(file_, record_size, record_capacity)


    @property
    def first_available_pid(self) -> RmPageId | None:
        """
        """
        page, header = RmFile.fetch_header(self.file_)
        pid_ = header.get('first_available_pid')
        page.unpin()
        if pid_ == int(RmFileHeader.NO_AVAILABLE_PAGE):
            return None
        return RmPageId(pid_)

    @first_available_pid.setter
    def first_available_pid(self, pid: RmPageId | None) -> None:
        if pid is None:
            pid = RmFileHeader.NO_AVAILABLE_PAGE
        page, header = RmFile.fetch_header(self.file_)
        header.set('first_available_pid', int(pid))
        header.commit()
        page.unpin()



    @assignment
    def get_page(self, pid: RmPageId) -> RmPage:
        """
        Raises:
            RmFileHeaderAccessError:
        """
        raise NotImplementedError




    @assignment
    def find_next_pid(self, pid: RmPageId) -> RmPageId | None:
        """
        """
        raise NotImplementedError




    @assignment
    def get_record(self, rid: RmRecordId) -> RmRecord:
        """
        Raises:
            RmInvalidSlotIdError:
            RmSlotUnoccupiedError:
        """
        raise NotImplementedError




    @assignment
    def insert_record(self, data: RmRecordData) -> RmRecordId:
        """
        Raises:
            RmInvalidSlotIdError:
            RmInvalidRecordDataSizeError:
        """
        raise NotImplementedError




    @assignment
    def delete_record(self, rid: RmRecordId) -> None:
        """
        Raises:
            RmInvalidSlotIdError:
            RmSlotUnoccupiedError:
        """
        raise NotImplementedError




    @assignment
    def update_record(self, rid: RmRecordId, data: RmRecordData) -> None:
        """
        Raises:
            RmInvalidSlotIdError:
            RmSlotUnoccupiedError:
            RmInvalidRecordDataSizeError:
        """
        raise NotImplementedError




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




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