from typing import (
Iterator,
)
from .error import (
RmUnoccupiedSlotError,
RmNoNextPageFoundError,
RmInvalidPredicateOffsetError,
)
from .predicate import (
RmPredicate,
)
from .record import (
RmRecordKey,
RmRecord,
)
from .page import (
RmSlotId,
RmPageId,
)
from .file import (
RmFileHeader,
RmFile,
)
__all__ = (
'RmFileScan',
)
[문서]
class RmFileScan:
"""
file scan을 나타내는 클래스입니다.
TODO: 추가 설명
Attributes:
file (RmFile):
scan할 file.
predicate (RmPredicate | None):
scan의 predicate.
cursor (RmRecordKey):
scan의 cursor.
"""
NO_NEXT_PAGE = RmPageId(-1)
"""
scan이 끝난 경우의 cursor의 page ID.
"""
INITIAL_PID = RmPageId(RmFileHeader.PID)
"""
초기 cursor의 page ID.
"""
INITIAL_SID = RmSlotId(0)
"""
초기 cursor의 slot ID.
"""
[문서]
def __init__(self,
file: RmFile,
predicate: RmPredicate | None = None,
) -> None:
"""
file scan을 초기화합니다.
TODO: 추가 설명
Args:
file (RmFile):
scan할 file.
predicate (RmPredicate | None):
scan의 predicate.
Raises:
RmInvalidPredicateOffsetError:
`predicate.offset` 이 유효하지 않은 경우.
"""
# validate predicate
if predicate is not None:
record_size = file.record_size
offset = predicate.offset
limit = predicate.domain.calc_size()
if offset < 0 or offset + limit > record_size:
raise RmInvalidPredicateOffsetError(
f"`predicate.offset`은 0 이상, {record_size - limit} 이하만 가능합니다 "
f"(현재 입력: {offset}).",
)
# init attributes
self.file = file
self.predicate = predicate
# init cursor
try:
pid = file.find_next_page(RmFileScan.INITIAL_PID).pid
except RmNoNextPageFoundError:
pid = RmFileScan.NO_NEXT_PAGE
self.cursor = RmRecordKey(
pid=pid,
sid=RmFileScan.INITIAL_SID,
)
[문서]
def next(self) -> RmRecord:
"""
file scan의 다음 record를 반환합니다.
TODO: 추가 설명
Returns:
RmRecord:
다음 record.
Raises:
StopIteration:
scan이 끝난 경우.
"""
raise NotImplementedError
[문서]
def __iter__(self) -> Iterator[RmRecord]:
"""
file scan을 iterator로 사용할 수 있게 합니다.
Yields:
RmRecord:
scan한 record.
"""
return self
[문서]
def __next__(self) -> RmRecord:
"""
file scan iterator의 다음 record를 반환합니다.
Returns:
RmRecord:
다음 record.
Raises:
StopIteration:
scan이 끝난 경우.
"""
return self.next()