from pathlib import Path
from dataclasses import dataclass
from collections.abc import Iterator
from bluebase.core import (
Value,
Ast,
AstSelect,
AstInsert,
AstDelete,
AstUpdate,
AstCreateTable,
AstCreateIndex,
AstDropTable,
AstDropIndex,
AstShowTables,
AstDescribeTable,
AstDumpTable,
AstExit,
dump,
)
from bluebase.pf import PfManager
from bluebase.rm import RmManager
from bluebase.ix import IxManager
from bluebase.sm import SmManager
from bluebase.ql import (
QlManager,
QlOp,
)
from bluebase.lp import LpManager
__all__ = (
'Result',
'BlueBase',
)
[docs]
@dataclass(frozen=True, slots=True)
class Result:
"""
An execution result.
Attributes:
query: the input query
op: the operation for the query
"""
query: str
op: QlOp
def __repr__(self) -> str: # pragma: no cover
return f"Result({self.query!r})"
[docs]
def __iter__(self) -> Iterator[tuple[Value, ...]]:
"""
Yield flattened rows from the operation.
"""
for row in self.op:
yield row.flatten()
[docs]
class BlueBase:
"""
The main API.
Open an existing database at the given path, or create a new one if it does not exist.
Attributes:
path: the path to the database folder
pf: a `PF <bluebase.pf>` manager
rm: a `RM <bluebase.rm>` manager
ix: a `IX <bluebase.ix>` manager
sm: a `SM <bluebase.sm>` manager
ql: a `QL <bluebase.ql>` manager
lp: a `LP <bluebase.lp>` manager
"""
path: Path
pf: PfManager
rm: RmManager
ix: IxManager
sm: SmManager
ql: QlManager
lp: LpManager
def __repr__(self) -> str: # pragma: no cover
return f"BlueBase({self.path!r})"
def __init__(self, path: Path) -> None:
self.path = path
self.pf = PfManager()
self.rm = RmManager(self.pf)
self.ix = IxManager(self.pf)
self.sm = SmManager(self.rm, self.ix)
self.ql = QlManager(self.sm)
self.ql.optim.push_file = True
self.ql.optim.push_index = True
self.ql.optim.loop_join = True
self.ql.optim.hash_join = True
self.ql.optim.index_join = True
self.ql.optim.first_leaf = True
self.lp = LpManager()
if self.path.exists():
self.sm.open_database(self.path)
else:
self.sm.create_database(self.path)
self.sm.open_database(self.path)
def _cli_only(self, ast: Ast) -> None:
raise ValueError(f"CLI-only feature ({dump(ast=ast.verbose())}).")
[docs]
def close(self) -> None:
"""
Close the database.
"""
self.sm.close_database()
[docs]
def execute(self, query: str) -> Result | None:
"""
Execute a query and return its result.
Raises:
ValueError: if a CLI-only feature is executed
NotImplementedError: if an unsupported feature is executed
"""
ast = self.lp.parse(query)
match ast:
case AstSelect():
return Result(query, self.ql.select(ast))
case AstInsert():
self.ql.insert(ast)
case AstDelete():
self.ql.delete(ast)
case AstUpdate():
self.ql.update(ast)
case AstCreateTable():
self.sm.create_table(ast.table.name, {
define.attr.name: define.domain
for define in ast.defines
})
case AstCreateIndex():
self.sm.create_index(ast.table.name, ast.attr.name)
case AstDropTable():
self.sm.drop_table(ast.table.name)
case AstDropIndex():
self.sm.drop_index(ast.table.name, ast.attr.name)
case AstExit():
self._cli_only(ast)
case AstShowTables():
self._cli_only(ast)
case AstDescribeTable():
self._cli_only(ast)
case AstDumpTable():
self._cli_only(ast)
case _: # pragma: no cover
raise NotImplementedError(f"Not implemented yet ({dump(
ast=ast.verbose(),
)}).")
return None