from datetime import datetime
from dataclasses import is_dataclass
from decimal import Decimal
from typing import Any, get_type_hints, Generator
from dateutil import parser
from lxml import etree
from lxml.objectify import (
ObjectifiedElement,
IntElement,
StringElement,
BoolElement,
FloatElement,
NumberElement,
)
from q2_sdk.hq.db.representation_row_base import RepresentationRowBase
[docs]
class ExtendedBase:
RAW_TYPE = str
@property
def text(self):
return str(self)
@property
def pyval(self):
return self.RAW_TYPE(self)
[docs]
class ExtendedString(str, ExtendedBase):
RAW_TYPE = str
[docs]
class ExtendedInt(int, ExtendedBase):
RAW_TYPE = int
[docs]
class ExtendedFloat(float, ExtendedBase):
RAW_TYPE = float
[docs]
class ExtendedDateTime(datetime, ExtendedBase):
@property
def pyval(self):
return parser.parse(self.text)
def __json__(self):
return self.pyval.isoformat()
[docs]
class ExtendedDecimal(Decimal, ExtendedBase):
@property
def pyval(self):
return Decimal(self.text)
def __json__(self):
return str(self.pyval)
# This can't follow the same pattern as the above since bool is not subclassable
[docs]
class ExtendedBool(ExtendedBase):
RAW_TYPE = bool
def __init__(self, value):
self._val = value
@property
def text(self):
return str(self._val)
@property
def pyval(self):
from q2_sdk.tools.utils import to_bool
return to_bool(self._val)
def __repr__(self):
return str(self.pyval)
def __str__(self):
return str(self.pyval)
def __eq__(self, other: object) -> bool:
if isinstance(other, bool):
return self.pyval == other
return super().__eq__(other)
def __json__(self):
return self.text
TYPE_MAP = {
IntElement: ExtendedInt,
StringElement: ExtendedString,
BoolElement: ExtendedBool,
FloatElement: ExtendedFloat,
NumberElement: ExtendedFloat,
int: ExtendedInt,
str: ExtendedString,
bool: ExtendedBool,
float: ExtendedFloat,
Decimal: ExtendedDecimal,
datetime: ExtendedDateTime,
}
[docs]
class TableRow:
"""
Returned from a DBObject call to the DB if setting ``RETURN_TABLE_OBJECTS_FROM_DB``
is set to ``True``. Allows for converting of LXML ObjectifiedElement.text to the
defined type in the RepresentationRow class for each DBObject. This
RepresentationRow should match the schema in the DB for the associated DB table.
Also accessing ``.text`` and ``.pyval`` are no longer necessary. Just accessing the
field name will return the correct value as the correct type. Also included are the
``__repr__`` and ``__str__`` methods that will show the string representation of the
xml underneath to make debugging easier.
"""
def __init__(
self, element: ObjectifiedElement, row_class: type[RepresentationRowBase]
) -> None:
object.__setattr__(self, "_element", element)
object.__setattr__(self, "_row_class", row_class)
object.__setattr__(self, "_current_row_index", 0)
object.__setattr__(self, "_items", {})
self._set_items()
@property
def headers(self) -> list[str]:
return list(object.__getattribute__(self, "_items").keys())
[docs]
def keys(self) -> list[str]:
return [x[0] for x in self.items()]
[docs]
def values(self) -> list[Any]:
return [x[1] for x in self.items()]
def _set_items(self):
_row_class = object.__getattribute__(self, "_row_class")
_items = object.__getattribute__(self, "_items")
if not _items:
headers = [attr for attr in dir(_row_class) if not attr.startswith("__")]
for header in headers:
_items[header] = self.__getattr__(header)
[docs]
def items(self) -> Generator[tuple[str, Any], None, None]:
_items = object.__getattribute__(self, "_items")
for key, value in _items.items():
yield key, value
def __setitem__(self, key: str, value: Any) -> None:
self._items[key] = value
self._element[key] = value
def __setattr__(self, key: str, value: Any) -> None:
self._items[key] = value
self._element[key] = value
def __getitem__(self, key: str) -> Any:
return self.__getattr__(key)
def __getattr__(self, item: str) -> Any:
_items = object.__getattribute__(self, "_items")
if item in _items:
return _items[item]
_row_class = object.__getattribute__(self, "_row_class")
_element = object.__getattribute__(self, "_element")
__class__ = object.__getattribute__(self, "__class__")
if hasattr(_row_class, item):
cell = getattr(_element, item, None)
if cell is not None and cell.text:
types = get_type_hints(_row_class)
_type = types[item]
if _type == datetime:
data = ExtendedDateTime.fromisoformat(
parser.parse(cell.text).isoformat()
)
elif is_dataclass(_type):
data = _type.from_xml(cell.text)
else:
data = TYPE_MAP[_type](cell.text)
return data
else:
return None
if hasattr(_element, item):
return getattr(_element, item)
raise AttributeError(f"'{__class__.__name__}' object has no attribute '{item}'")
def __contains__(self, item) -> bool:
return item in self.headers
def __iter__(self) -> "TableRow":
object.__setattr__(self, "_current_row_index", 0)
return self
def __next__(self) -> Any:
_current_row_index = object.__getattribute__(self, "_current_row_index")
if _current_row_index < len(self):
value = self.__getattr__(self.headers[_current_row_index])
object.__setattr__(self, "_current_row_index", _current_row_index + 1)
return value
else:
raise StopIteration
def __len__(self) -> int:
return len(self.headers)
def _to_str(self) -> str:
_element = object.__getattribute__(self, "_element")
return etree.tostring(_element).decode("utf-8")
def __str__(self) -> str:
return self._to_str()
def __repr__(self) -> str:
return f"{self.__class__.__name__}('{self._to_str()}')"