from pathlib import Path
from bluebase.core import dump
from .error import (
PfInvalidPageSizeError,
PfInvalidCacheCapacityError,
PfManagedFileCreateError,
PfManagedFileDestroyError,
PfUnmanagedFileCloseError,
)
from .type import PfFileId
from .replacer import (
PfReplacer,
PfLruReplacer,
)
from .cache import PfPageCache
from .page import PfPageHeader
from .file import PfFile
__all__ = (
'PfManager',
)
[docs]
class PfManager:
"""
A manager for the `PF <bluebase.pf>` component.
The manager manages files and pages in the database.
Note that the page cache covers multiple files.
Attributes:
page_size: the size of a page
cache_capacity: the capacity of the page cache
replacer: the replacement policy of the page cache
cache: the page cache object
fid_counter: the container for managing file IDs
Raises:
PfInvalidPageSizeError: if the page size is invalid
PfInvalidCacheCapacityError: if the cache capacity is invalid
"""
page_size: int
cache_capacity: int
replacer: PfReplacer
cache: PfPageCache
fid_counter: dict[PfFileId, int]
def __repr__(self) -> str: # pragma: no cover
return f"PfManager({dump(
page_size=self.page_size,
cache_capacity=self.cache_capacity,
replacer=self.replacer,
)})"
def __init__(self,
*,
page_size: int = 4096,
cache_capacity: int = 40,
replacer: PfReplacer | None = None,
) -> None:
if page_size <= PfPageHeader.size:
raise PfInvalidPageSizeError(
size=page_size,
header=PfPageHeader.size,
)
if cache_capacity <= 0:
raise PfInvalidCacheCapacityError(capacity=cache_capacity)
self.page_size = page_size
self.cache_capacity = cache_capacity
self.replacer = replacer or PfLruReplacer()
self.cache = PfPageCache(
page_size=page_size,
capacity=cache_capacity,
replacer=self.replacer,
)
self.fid_counter: dict[PfFileId, int] = {}
[docs]
def create_file(self, path: Path) -> None:
"""
Create a new file.
Raises:
PfManagedFileCreateError: if the file is already managed
"""
path.touch(exist_ok=False)
fid = PfFileId.new(path)
if fid in self.fid_counter:
raise PfManagedFileCreateError(path=path)
file = PfFile.new(fid, self.cache)
file.persist()
[docs]
def destroy_file(self, path: Path) -> None:
"""
Destroy a file.
Raises:
PfManagedFileDestroyError: if the file is still being managed
"""
fid = PfFileId.new(path)
if fid in self.fid_counter:
raise PfManagedFileDestroyError(path=path)
path.unlink()
[docs]
def open_file(self, path: Path) -> PfFile:
"""
Open a file.
"""
fid = PfFileId.new(path)
if fid not in self.fid_counter:
self.fid_counter[fid] = 0
self.fid_counter[fid] += 1
file = PfFile.open(fid, self.cache)
return file
[docs]
def close_file(self, file: PfFile) -> None:
"""
Close a file.
Raises:
PfUnmanagedFileCloseError: if the file is not managed
"""
if file.fid not in self.fid_counter:
raise PfUnmanagedFileCloseError(file=file)
self.fid_counter[file.fid] -= 1
if self.fid_counter[file.fid] == 0:
file.flush()
file.stream.close()
del self.fid_counter[file.fid]