from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .manager import QlManager
from abc import (
ABC,
abstractmethod,
)
from dataclasses import (
field,
dataclass,
)
from collections.abc import Iterator
from bluebase.core import (
Value,
Domain,
Comp,
Predicate,
)
from bluebase.rm import RmRecord
from bluebase.sm import (
SmTableInfo,
SmAttrInfo,
)
__all__ = (
'QlSchema',
'QlContext',
'QlAttr',
'QlRow',
'QlCond',
'QlAttrCond',
'QlValueCond',
'QlAssign',
'QlAttrAssign',
'QlValueAssign',
)
[docs]
@dataclass(frozen=True, slots=True)
class QlSchema:
"""
Attributes:
table:
attrs:
"""
table: SmTableInfo
attrs: dict[str, SmAttrInfo]
def verbose(self) -> str: # pragma: no cover
return f"{self.table.name}:[{','.join(self.attrs.keys())}]"
[docs]
@dataclass(frozen=True, slots=True)
class QlContext:
"""
Attributes:
manager:
schemas:
"""
manager: 'QlManager'
schemas: dict[str, QlSchema]
def verbose(self) -> str: # pragma: no cover
return f"[{','.join(self.schemas.keys())}]"
[docs]
@dataclass(frozen=True, slots=True)
class QlAttr:
"""
Attributes:
ref:
name:
domain:
"""
ref: str
name: str
domain: Domain = field(compare=False)
def verbose(self) -> str: # pragma: no cover
return f"{self.ref}.{self.name}"
[docs]
@dataclass(frozen=True, slots=True)
class QlRow:
"""
Attributes:
record:
lookup:
"""
record: RmRecord | None
lookup: dict[QlAttr, Value]
[docs]
def __or__(self, other: 'QlRow') -> 'QlRow':
"""
"""
return QlRow(None, {**self.lookup, **other.lookup})
[docs]
def __getitem__(self, attr: QlAttr) -> Value:
"""
"""
return self.lookup[attr]
[docs]
def items(self) -> Iterator[tuple[QlAttr, Value]]:
"""
"""
return iter(self.lookup.items())
[docs]
def restrict(self, attrs: list[QlAttr]) -> 'QlRow':
"""
"""
return QlRow(self.record, {attr: self.lookup[attr] for attr in attrs})
[docs]
def flatten(self) -> tuple[Value, ...]:
"""
"""
return tuple(self.lookup.values())
[docs]
@dataclass(frozen=True, slots=True)
class QlCond(ABC):
"""
Attributes:
left_attr:
comp:
"""
left_attr: QlAttr
comp: Comp
@property
def key(self) -> str:
"""
"""
return self.verbose()
[docs]
@abstractmethod
def get_predicate(self, row: QlRow) -> Predicate: # pragma: no cover
"""
"""
pass
[docs]
def evaluate(self, row: QlRow) -> bool:
"""
"""
predicate = self.get_predicate(row)
return predicate.evaluate(row.lookup[self.left_attr])
@abstractmethod
def verbose(self) -> str: # pragma: no cover
pass
[docs]
@dataclass(frozen=True, slots=True)
class QlAttrCond(QlCond):
"""
Attributes:
right_attr:
"""
right_attr: QlAttr
[docs]
def get_predicate(self, row: QlRow) -> Predicate:
return Predicate(
name=self.left_attr.name,
comp=self.comp,
value=row.lookup[self.right_attr],
)
def verbose(self) -> str: # pragma: no cover
return ''.join([
f"{self.left_attr.ref}.{self.left_attr.name}",
self.comp.value,
f"{self.right_attr.ref}.{self.right_attr.name}",
])
[docs]
@dataclass(frozen=True, slots=True)
class QlValueCond(QlCond):
"""
Attributes:
right_value:
"""
right_value: Value
[docs]
def get_predicate(self, row: QlRow) -> Predicate:
return Predicate(
name=self.left_attr.name,
comp=self.comp,
value=self.right_value,
)
def verbose(self) -> str: # pragma: no cover
return ''.join([
f"{self.left_attr.ref}.{self.left_attr.name}",
self.comp.value,
f"{self.right_value}",
])
[docs]
@dataclass(frozen=True, slots=True)
class QlAssign(ABC):
"""
Attributes:
left_attr:
"""
left_attr: QlAttr
@property
def key(self) -> str:
"""
"""
return self.left_attr.name
@abstractmethod
def verbose(self) -> str: # pragma: no cover
pass
[docs]
@dataclass(frozen=True, slots=True)
class QlAttrAssign(QlAssign):
"""
Attributes:
right_attr:
"""
right_attr: QlAttr
def verbose(self) -> str: # pragma: no cover
return ''.join([
f"{self.left_attr.ref}.{self.left_attr.name}",
"=",
f"{self.right_attr.ref}.{self.right_attr.name}",
])
[docs]
@dataclass(frozen=True, slots=True)
class QlValueAssign(QlAssign):
"""
Attributes:
right_value:
"""
right_value: Value
def verbose(self) -> str: # pragma: no cover
return ''.join([
f"{self.left_attr.ref}.{self.left_attr.name}",
"=",
f"{self.right_value}",
])