Source code for bluebase.sm.manager

import sys

from typing import (
    Any,
    TextIO,
    ClassVar,
)

from pathlib import Path

from bluebase.core import (
    Comp,
    Domain,
    Printer,
    dump,
    optional,
    assignment,
)
from bluebase.rm import (
    RmManager,
    RmFile,
    RmRecordData,
    RmRecord,
    RmPredicate,
    RmFileScan,
)
from bluebase.ix import (
    IxManager,
    IxEntry,
)

from .info import (
    SmTableInfo,
    SmAttrInfo,
)
from .error import (
    SmInvalidDatabasePathError,
    SmOpenedDatabaseOpenError,
    SmClosedDatabaseAccessError,
    SmNotAllowedTableNameError,
    SmInvalidTableNameError,
    SmTableAlreadyExistsError,
    SmTableNotFound,
    SmInvalidAttrNameError,
    SmAttrNotFoundError,
    SmInvalidSchemaSizeError,
    SmIndexAlreadyExistsError,
    SmIndexNotFoundError,
)


__all__ = (
    'SmManager',
)




class SmManager:
    """
    A manager for the `SM <bluebase.sm>` component.

    Attributes:
        rm:
        ix:
        opened:
        path:
        table_cat:
        attr_cat:
    """

    TABLE_CAT_NAME: ClassVar[str] = 'table_cat'
    """table catalog 이름을 나타내는 상수."""

    ATTR_CAT_NAME: ClassVar[str] = 'attr_cat'
    """attr catalog 이름을 나타내는 상수."""

    MAX_SCHEMA_SIZE: ClassVar[int] = 40
    """schema의 최대 크기를 나타내는 상수."""

    rm: RmManager
    ix: IxManager
    opened: bool
    path: Path | None
    table_cat: RmFile | None
    attr_cat: RmFile | None

    def __repr__(self) -> str:  # pragma: no cover
        return f"SmManager({dump(path=self.path)})"

    def __init__(self, rm: RmManager, ix: IxManager) -> None:
        self.rm = rm
        self.ix = ix
        self.opened = False
        self.path = None
        self.table_cat = None
        self.attr_cat = None



    def validate_opened(self) -> None:
        """
        """
        if not self.opened:
            raise SmClosedDatabaseAccessError(manager=self)




    def validate_table_name(self, table_name: str) -> None:
        """
        """
        if table_name in (SmManager.TABLE_CAT_NAME, SmManager.ATTR_CAT_NAME):
            raise SmNotAllowedTableNameError(table=table_name)
        length = len(table_name)
        if length < 1 or length > SmTableInfo.MAX_TABLE_LENGTH:
            raise SmInvalidTableNameError(
                table=table_name,
                min=1,
                max=SmTableInfo.MAX_TABLE_LENGTH,
            )




    def validate_attr_name(self, attr_name: str) -> None:
        """
        """
        if len(attr_name) < 1 or len(attr_name) > SmAttrInfo.MAX_ATTR_LENGTH:
            raise SmInvalidAttrNameError(
                attr=attr_name,
                min=1,
                max=SmAttrInfo.MAX_ATTR_LENGTH,
            )




    def validate_schema(self, schema: dict[str, Domain]) -> None:
        """
        """
        if len(schema) < 1 or len(schema) > SmManager.MAX_SCHEMA_SIZE:
            raise SmInvalidSchemaSizeError(
                schema=len(schema),
                min=1,
                max=SmManager.MAX_SCHEMA_SIZE,
            )
        for attr_name in schema:
            self.validate_attr_name(attr_name)




    @optional
    def _search_table(self, table_name: str) -> RmRecord | None:
        """
        Hint:
            이런 함수가 있으면 편하다!
        """
        raise NotImplementedError




    @optional
    def _search_attrs(self, table_name: str) -> list[RmRecord]:
        """
        Hint:
            이런 함수가 있으면 편하다!
        """
        raise NotImplementedError




    def get_table_path(self, table_name: str) -> Path:
        """
        """
        assert self.path is not None
        return self.path / f'{table_name}.table'




    @assignment
    def get_table_info(self, table_name: str) -> SmTableInfo:
        """
        Raises:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def get_attr_infos(self, table_name: str) -> list[SmAttrInfo]:
        """
        Raises:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def create_database(self, path: Path) -> None:
        """
        Raises:
            SmOpenedDatabaseOpenError:
            SmInvalidDatabasePathError:
        """
        raise NotImplementedError




    @assignment
    def open_database(self, path: Path) -> None:
        """
        Raises:
            SmInvalidDatabasePathError:
            SmOpenedDatabaseOpenError:
        """
        raise NotImplementedError




    @assignment
    def close_database(self) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
        """
        raise NotImplementedError




    @assignment
    def create_table(self, table_name: str, schema: dict[str, Domain]) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmNotAllowedTableNameError:
            SmInvalidTableNameError:
            SmInvalidSchemaSizeError:
            SmTableAlreadyExistsError:
        """
        raise NotImplementedError




    @assignment
    def drop_table(self, table_name: str) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmNotAllowedTableNameError:
            SmInvalidTableNameError:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def create_index(self, table_name: str, attr_name: str) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmNotAllowedTableNameError:
            SmInvalidTableNameError:
            SmInvalidAttrNameError:
            SmTableNotFound:
            SmAttrNotFoundError:
            SmIndexAlreadyExistsError:
        """
        raise NotImplementedError




    @assignment
    def drop_index(self, table_name: str, attr_name: str) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmNotAllowedTableNameError:
            SmInvalidTableNameError:
            SmInvalidAttrNameError:
            SmTableNotFound:
            SmAttrNotFoundError:
            SmIndexNotFoundError:
        """
        raise NotImplementedError




    @assignment
    def bulk_load(self, table_name: str, rows: list[tuple[Any, ...]]) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmNotAllowedTableNameError:
            SmInvalidTableNameError:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def show_tables(self, stream: TextIO | None = None) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def describe_table(self,
                       table_name: str,
                       stream: TextIO | None = None,
                       ) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmTableNotFound:
        """
        raise NotImplementedError




    @assignment
    def dump_table(self, table_name: str, stream: TextIO | None = None) -> None:
        """
        Raises:
            SmClosedDatabaseAccessError:
            SmTableNotFound:
        """
        raise NotImplementedError