pf.file의 소스 코드

from typing import (
    BinaryIO,
    TYPE_CHECKING,
)
if TYPE_CHECKING:
    from .buffer import (
        PfBufferManager,
    )

from pathlib import (
    Path,
)

from .error import (
    PfInvalidPidError,
    PfNoNextAllocatedPageFoundError,
    PfDisposingPinnedPageError,
    PfAlreadyDisposedPageError,
)
from .layout import (
    PfLayoutField,
    PfLayoutView,
)
from .page import (
    PfPageId,
    PfPageHeader,
    PfPage,
)


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




class PfFileId(int):
    """
    file의 ID를 나타내는 클래스입니다.
    """

    pass





class PfFileHeader(PfLayoutView):
    """
    file의 header를 나타내는 클래스입니다.

    file header는 0번째 페이지에 저장됩니다.
    page content의 첫 8바이트를 사용하는 layout view입니다.
    layout view의 property를 사용 할 때는 항상 해당 page가 pin 되어있어야 합니다.

    Attributes:
        page_count (int):
            file에 존재하는 page 수 (free page 포함).
        first_free_pid (PfPageId):
            file에 있는 가장 첫번째 free page의 ID.
            `NO_FREE_PAGE` 인 경우는 file에 free page가 없다는 뜻입니다.
    """

    PID = PfPageId(0)
    """
    file header가 들어있는 page의 ID.
    """

    NO_FREE_PAGE = PfPageId(-1)
    """
    file에 free page가 없다는 뜻을 나타내는 상수.
    """

    _fields = (
        PfLayoutField('page_count', '<i'),
        PfLayoutField('first_free_pid', '<i'),
    )

    @property
    def page_count(self) -> int:
        return int(self.get('page_count'))

    @page_count.setter
    def page_count(self, count: int) -> None:
        self.set('page_count', int(count))

    @property
    def first_free_pid(self) -> PfPageId:
        return PfPageId(self.get('first_free_pid'))

    @first_free_pid.setter
    def first_free_pid(self, pid: PfPageId) -> None:
        self.set('first_free_pid', int(pid))





class PfFile:
    """
    file을 나타내는 클래스입니다.

    Attributes:
        fid (PfFileId):
            file의 ID.
        path (Path):
            file의 경로.
        stream (BinaryIO):
            file의 I/O stream.
        _buffer_manager (PfBufferManager):
            이 파일을 관리하는 buffer manager.
    """



    def __init__(self,
                 fid: PfFileId,
                 path: Path,
                 stream: BinaryIO,
                 buffer_manager: 'PfBufferManager',
                 ) -> None:
        """
        file을 초기화합니다.

        Args:
            fid (PfFileId):
                file의 ID.
            path (Path):
                file의 경로.
            stream (BinaryIO):
                file의 I/O stream.
            buffer_manager (PfBufferManager):
                이 파일을 관리하는 buffer manager.
        """
        self.fid = fid
        self.path = path
        self.stream = stream
        self._buffer_manager = buffer_manager




    def get_page(self, pid: PfPageId) -> PfPage:
        """
        page를 가져옵니다.
        free page도 가져올 수 있습니다.

        buffer manager 의 `fetch_page` 메서드를 사용합니다.

        Args:
            pid (PfPageId):
                대상 page의 ID.

        Returns:
            PfPage:
                가져온 page.

        Raises:
            PfInvalidPidError:
                `pid` 가 유효하지 않은 경우.
        """
        raise NotImplementedError




    def find_next_allocated_page(self, pid: PfPageId) -> PfPage:
        """
        file에 있는, `pid` 다음의 allocate된 page를 찾습니다.
        free page는 건너뜁니다.

        Args:
            pid (PfPageId):
                대상 page의 ID.

        Returns:
            PfPage:
                `pid` 다음에 있는 allocate된 page.

        Raises:
            PfInvalidPidError:
                `pid` 가 유효하지 않은 경우.
            PfNoNextAllocatedPageFoundError:
                `pid` 다음에 있는 allocate된 page를 찾을 수 없는 경우.
        """
        raise NotImplementedError




    def allocate_page(self) -> PfPage:
        """
        page를 allocate합니다.

        free page가 없으면, 새로운 page를 allocate합니다.
        buffer manager 의 `allocate_page` 메서드를 사용하며,
        file header의 `page_count` 를 증가시킵니다.

        free page가 있으면, 해당 page를 재활용 합니다.
        buffer manager 의 `fetch_page` 메서드를 사용하며,
        file header의 `first_free_pid` 를 재활용할 page header의 `next_free_pid` 로 업데이트 하고,
        page header의 `next_free_pid` 를 `NOT_IN_FREE_PAGE_LIST` 로 업데이트 합니다.
        page data는 page의 `clear_content` 메서드를 사용하여 초기화합니다.

        Returns:
            PfPage:
                allocate된 page.
        """
        raise NotImplementedError




    def dispose_page(self, pid: PfPageId) -> None:
        """
        page를 해제하여 free page로 만듭니다.

        page header의 `next_free_pid` 를 file header의 `first_free_pid` 로 업데이트 하고,
        file header의 `first_free_pid` 를 해제할 page의 ID로 업데이트 합니다.

        Args:
            pid (PfPageId):
                대상 page의 ID.

        Raises:
            PfInvalidPidError:
                `pid` 가 file header page의 ID거나 유효하지 않은 경우.
            PfDisposingPinnedPageError:
                page가 pin 되어있어서 해제할 수 없는 경우.
            PfAlreadyDisposedPageError:
                page가 이미 해제된 상태인 경우.
        """
        raise NotImplementedError




    def flush(self) -> int:
        """
        file을 flush 합니다.

        buffer manager 의 `flush` 메서드를 사용합니다.

        Returns:
            int:
                write가 발생한 page의 수.

        Raises:
            PfFlushingPinnedPageError:
                page가 pin 되어있어서 flush할 수 없는 경우.
        """
        raise NotImplementedError