from dataclasses import dataclass
from collections.abc import Iterator
from bluebase.core import (
Domain,
Predicate,
dump,
optional,
assignment,
)
from .error import RmRecordOutOfRangeError
from .type import (
RmSlotId,
RmPageId,
RmRecord,
)
from .file import (
RmFileHeader,
RmFile,
)
__all__ = (
'RmPredicate',
'RmScanCursor',
'RmFileScan',
)
[docs]
@dataclass(frozen=True, slots=True)
class RmPredicate(Predicate):
"""
file scan을 위한 predicate를 나타내는 `dataclass` 클래스입니다.
Attributes:
domain:
offset:
"""
domain: Domain
offset: int
def __post_init__(self) -> None:
self.domain.validate(self.value)
[docs]
def slice(self) -> slice:
"""
"""
return slice(
self.offset,
self.offset + self.domain.size,
)
[docs]
def validate_offset(self, record_size: int) -> None:
"""
Raises:
RmRecordOutOfRangeError:
"""
slice = self.slice()
if slice.start < 0 or slice.stop > record_size:
raise RmRecordOutOfRangeError(
record=record_size,
offset=self.offset,
limit=self.domain.size,
)
[docs]
def evaluate_record(self, record: RmRecord) -> bool:
"""
Raises:
RmRecordOutOfRangeError:
"""
self.validate_offset(len(record.data))
data = record.data[self.slice()]
value = self.domain.decode(data)
return self.evaluate(value)
[docs]
@dataclass(slots=True)
class RmScanCursor:
"""
Attributes:
pid:
sid:
"""
pid: RmPageId
sid: RmSlotId
[docs]
class RmFileScan:
"""
Attributes:
file:
predicate:
cursor:
Raises:
RmRecordOutOfRangeError:
"""
file: RmFile
predicate: RmPredicate | None
cursor: RmScanCursor | None
def __repr__(self) -> str: # pragma: no cover
return f"RmFileScan({dump(
file=self.file,
predicate=self.predicate,
)})"
def __init__(self,
file: RmFile,
predicate: RmPredicate | None = None,
) -> None:
if predicate is not None:
predicate.validate_offset(file.record_size)
self.file = file
self.predicate = predicate
self.cursor = None
[docs]
def __iter__(self) -> Iterator[RmRecord]:
"""
file scan을 순회하는 iterator를 반환합니다.
"""
return self
[docs]
@assignment
def __next__(self) -> RmRecord:
"""
Raises:
StopIteration:
"""
raise NotImplementedError
[docs]
@optional
def _init_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _advance_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _fetch_record(self) -> RmRecord | None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _match_record(self, record: RmRecord) -> bool:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError