OrmSerializer¶
OrmSerializer(
table_data,
table_map,
result_set,
is_array,
depth,
load_paths=None,
load_options=(),
)
Bases: Generic[SerializedType]
Generate Python models from a table map and result set.
Source code in ormdantic/serializer.py
def __init__(
self,
table_data: OrmTable[Any],
table_map: Map,
result_set: Any,
is_array: bool,
depth: int,
load_paths: tuple[str, ...] | None = None,
load_options: tuple[LoaderOption, ...] = (),
) -> None:
self._table_data = table_data
self._table_map = table_map
self._result_set = result_set
self._is_array = is_array
self._depth = depth
self._load_paths = load_paths
self._load_options = load_options
self._identity_map: dict[tuple[type[BaseModel], Any], BaseModel] = {}
self._building_identities: set[tuple[type[BaseModel], Any]] = set()
result_schema = (
self._get_path_result_schema(
table_data, self._path_tree(load_paths), is_array
)
if load_paths is not None
else self._get_result_schema(table_data, depth, is_array)
)
self._result_schema = ResultSchema(
is_array=is_array,
references={table_data.tablename: result_schema},
)
self._columns = [it[0] for it in self._result_set.cursor.description]
self._return_dict: dict[str, Any] = {}
deserialize()
Deserialize the result set into Python models.
Source code in ormdantic/serializer.py
def deserialize(self) -> SerializedType:
"""Deserialize the result set into Python models."""
if self._depth <= 0 and self._load_paths is None:
return self._deserialize_flat()
self._return_dict = (
hydrate_joined_payload(
columns=self._columns,
rows=[tuple(row) for row in self._result_set],
path_pks=self._path_pks(self._result_schema),
array_paths=self._array_paths(self._result_schema),
)
or {}
)
if not self._return_dict:
return None # type: ignore
root_schema = self._result_schema.references[self._table_data.tablename]
prepared = self._prep_result(self._return_dict, self._result_schema)[
self._table_data.tablename
]
if self._result_schema.is_array:
result = [
self._build_model(record, root_schema, cache_identity=False)
for record in prepared
]
return cast(SerializedType, self._apply_loader_options(result))
model = self._build_model(prepared, root_schema, cache_identity=False)
return cast(SerializedType, self._apply_loader_options(model))