Source code for bluebase.rm.page

from typing import (
    Self,
    ClassVar,
)

from bluebase.core import (
    LayoutField,
    LayoutMixin,
    dump,
    assignment,
)
from bluebase.pf import (
    PfPageView,
    PfPageData,
    PfPage,
)

from .error import (
    RmInvalidSlotIdError,
    RmSlotUnoccupiedError,
    RmInvalidRecordDataSizeError,
)
from .type import (
    RmSlotId,
    RmPageId,
    RmRecordId,
    RmRecordData,
    RmRecord,
)


__all__ = (
    'RmPageHeader',
    'RmPageBitmap',
    'RmPage',
)




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

    .. rubric:: Layout

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

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

    NO_NEXT_AVAILABLE_PAGE: ClassVar[RmPageId] = RmPageId(-1)
    """다음 available page가 없을 경우 사용하는 상수."""

    fields = (
        LayoutField('next_available_pid', 'i'),
    )





class RmPageBitmap(PfPageView):
    """
    Attributes:
        capacity:
    """

    capacity: int

    def __repr__(self) -> str:  # pragma: no cover
        return f"RmPageBitmap({dump(
            page=self.page,
            offset=self.offset,
            limit=self.limit,
            capacity=self.capacity,
        )})"

    def __init__(self, page: PfPage, *, offset: int, capacity: int) -> None:
        super().__init__(
            page,
            offset=offset,
            limit=RmPageBitmap.calc_size(capacity=capacity),
        )
        self.capacity = capacity



    @staticmethod
    def calc_size(*, capacity: int) -> int:
        """
        """
        return (capacity + 7) // 8




    @assignment
    def find_first_zero(self) -> int:
        """
        """
        raise NotImplementedError




    @assignment
    def set_bit(self, position: int) -> None:
        """
        """
        raise NotImplementedError




    @assignment
    def clear_bit(self, position: int) -> None:
        """
        """
        raise NotImplementedError




    @assignment
    def check_bit(self, position: int) -> bool:
        """
        """
        raise NotImplementedError




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




    @assignment
    def is_empty(self) -> bool:
        """
        """
        raise NotImplementedError




    @assignment
    def is_full(self) -> bool:
        """
        """
        raise NotImplementedError






class RmPage:
    """
    Attributes:
        pid:
        page_:
        record_size:
        record_capacity:
    """

    pid: RmPageId
    page_: PfPage
    record_size: int
    record_capacity: int

    def __repr__(self) -> str:  # pragma: no cover
        return f"RmPage({self.pid})"

    def __init__(self,
                 page_: PfPage,
                 record_size: int,
                 record_capacity: int,
                 ) -> None:
        self.pid = RmPageId(page_.pid)
        self.record_size = record_size
        self.record_capacity = record_capacity
        self.page_ = page_



    @classmethod
    def new(cls,
            page_: PfPage,
            record_size: int,
            record_capacity: int,
            ) -> Self:
        """
        """
        page = cls(page_, record_size, record_capacity)
        header = page.get_header()
        header.set(
            'next_available_pid',
            int(RmPageHeader.NO_NEXT_AVAILABLE_PAGE),
        )
        header.commit()
        bitmap = page.get_bitmap()
        bitmap.reset()
        bitmap.commit()
        return page


    @property
    def header_offset(self) -> int:
        """
        """
        return self.page_.payload_offset

    @property
    def header_size(self) -> int:
        """
        """
        return RmPageHeader.size

    @property
    def bitmap_offset(self) -> int:
        """
        """
        return self.header_offset + self.header_size

    @property
    def bitmap_size(self) -> int:
        """
        """
        return RmPageBitmap.calc_size(capacity=self.record_capacity)

    @property
    def payload_offset(self) -> int:
        """
        """
        return self.bitmap_offset + self.bitmap_size

    @property
    def last_sid(self) -> RmSlotId:
        """
        """
        return RmSlotId(self.record_capacity - 1)

    @property
    def next_available_pid(self) -> RmPageId | None:
        """
        """
        header = self.get_header()
        pid_ = header.get('next_available_pid')
        if pid_ == int(RmPageHeader.NO_NEXT_AVAILABLE_PAGE):
            return None
        return RmPageId(pid_)

    @next_available_pid.setter
    def next_available_pid(self, pid: RmPageId | None) -> None:
        if pid is None:
            pid = RmPageHeader.NO_NEXT_AVAILABLE_PAGE
        header = self.get_header()
        header.set('next_available_pid', int(pid))
        header.commit()



    def validate_sid(self, sid: RmSlotId) -> None:
        """
        """
        if int(sid) < 0 or int(sid) >= self.record_capacity:
            raise RmInvalidSlotIdError(
                page=self,
                sid=sid,
                record_capacity=self.record_capacity,
            )




    def validate_slot_occupied(self, sid: RmSlotId) -> None:
        """
        """
        bitmap = self.get_bitmap()
        if not bitmap.check_bit(int(sid)):
            raise RmSlotUnoccupiedError(page=self, sid=sid)




    def validate_slot_unoccupied(self, sid: RmSlotId) -> None:
        """
        """
        bitmap = self.get_bitmap()
        assert bitmap.check_bit(int(sid)) is False




    def get_header(self) -> RmPageHeader:
        """
        """
        return RmPageHeader(
            self.page_,
            offset=self.header_offset,
            limit=self.header_size,
        )




    def get_bitmap(self) -> RmPageBitmap:
        """
        """
        return RmPageBitmap(
            self.page_,
            offset=self.bitmap_offset,
            capacity=self.record_capacity,
        )




    def get_record_offset(self, sid: RmSlotId) -> int:
        """
        """
        assert 0 <= int(sid) < self.record_capacity
        return self.payload_offset + int(sid) * self.record_size




    @assignment
    def is_empty(self) -> bool:
        """
        """
        raise NotImplementedError




    @assignment
    def is_full(self) -> bool:
        """
        """
        raise NotImplementedError




    @assignment
    def find_first_free_slot(self) -> RmSlotId:
        """
        """
        raise NotImplementedError




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




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




    @assignment
    def set_record(self,
                   sid: RmSlotId,
                   data: RmRecordData,
                   *,
                   new: bool = False,
                   ) -> None:
        """
        Raises:
            RmInvalidSlotIdError:
            RmSlotUnoccupiedError:
            RmInvalidRecordDataSizeError:
        """
        raise NotImplementedError




    @assignment
    def clear_slot(self, sid: RmSlotId) -> None:
        """
        Raises:
            RmInvalidSlotIdError:
            RmSlotUnoccupiedError:
        """
        raise NotImplementedError




    @assignment
    def check_slot(self, sid: RmSlotId) -> bool:
        """
        Raises:
            RmInvalidSlotIdError:
        """
        raise NotImplementedError