Repository URL to install this package:
|
Version:
4.5.4.dev1 ▾
|
import typing as t
from sarus_data_spec.constants import ARRAY_VALUES, LIST_VALUES, OPTIONAL_VALUE
from sarus_data_spec.path import straight_path
import sarus_data_spec.typing as st
def fk_visitor(
_type: st.Type, curr_path: t.Optional[st.Path] = None
) -> t.Dict[st.Path, st.Path]:
"""Go over the type and returns a dict pointing/pointed"""
class AddFK(st.TypeVisitor):
result: t.Dict[st.Path, st.Path] = {}
def Id(
self,
unique: bool,
reference: t.Optional[st.Path] = None,
base: t.Optional[st.IdBase] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
if (reference is not None) and (curr_path is not None):
self.result = {curr_path: reference}
def Struct(
self,
fields: t.Mapping[str, st.Type],
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
result: t.Dict[st.Path, st.Path] = {}
if curr_path is None:
for item_name, item_type in fields.items():
additional_paths = fk_visitor(
item_type,
curr_path=straight_path([item_name]),
)
result = {**result, **additional_paths}
else:
for item_name, item_type in fields.items():
additional_paths = fk_visitor(
item_type,
curr_path=straight_path(
[
*(
element
for element in curr_path.to_strings_list()[
0
]
),
item_name,
]
),
)
result = {**result, **additional_paths}
self.result = result
def Union(
self,
fields: t.Mapping[str, st.Type],
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
result: t.Dict[st.Path, st.Path] = {}
if curr_path is None:
for item_name, item_type in fields.items():
additional_paths = fk_visitor(
item_type,
curr_path=straight_path([item_name]),
)
result = {**result, **additional_paths}
else:
for item_name, item_type in fields.items():
additional_paths = fk_visitor(
item_type,
curr_path=straight_path(
[
*(
element
for element in curr_path.to_strings_list()[
0
]
),
item_name,
]
),
)
result = {**result, **additional_paths}
self.result = result
def Optional(
self,
type: st.Type,
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
if curr_path is None:
additional_paths = fk_visitor(
type,
curr_path=straight_path([OPTIONAL_VALUE]),
)
else:
additional_paths = fk_visitor(
type,
curr_path=straight_path(
[
*(
element
for element in curr_path.to_strings_list()[0]
),
OPTIONAL_VALUE,
]
),
)
self.result = {**self.result, **additional_paths}
def List(
self,
type: st.Type,
max_size: int,
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
if curr_path is None:
additional_paths = fk_visitor(
type,
curr_path=straight_path([LIST_VALUES]),
)
else:
additional_paths = fk_visitor(
type,
curr_path=straight_path(
[
*(
element
for element in curr_path.to_strings_list()[0]
),
LIST_VALUES,
]
),
)
self.result = {**self.result, **additional_paths}
def Array(
self,
type: st.Type,
shape: t.Tuple[int, ...],
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
if curr_path is None:
additional_paths = fk_visitor(
type,
curr_path=straight_path([ARRAY_VALUES]),
)
else:
additional_paths = fk_visitor(
type,
curr_path=straight_path(
[
*(
element
for element in curr_path.to_strings_list()[0]
),
ARRAY_VALUES,
]
),
)
self.result = {**self.result, **additional_paths}
def Boolean(
self, properties: t.Optional[t.Mapping[str, str]] = None
) -> None:
pass
def Bytes(
self, properties: t.Optional[t.Mapping[str, str]] = None
) -> None:
pass
def Unit(
self, properties: t.Optional[t.Mapping[str, str]] = None
) -> None:
pass
def Constrained(
self,
type: st.Type,
constraint: st.Predicate,
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Date(
self,
format: str,
min: str,
max: str,
base: st.DateBase,
possible_values: t.Iterable[str],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Time(
self,
format: str,
min: str,
max: str,
base: st.TimeBase,
possible_values: t.Iterable[str],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Datetime(
self,
format: str,
min: str,
max: str,
base: st.DatetimeBase,
possible_values: t.Iterable[str],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Duration(
self,
unit: str,
min: int,
max: int,
possible_values: t.Iterable[int],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Enum(
self,
name: str,
name_values: t.Sequence[t.Tuple[str, int]],
ordered: bool,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Text(
self,
encoding: str,
possible_values: t.Iterable[str],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Hypothesis(
self,
*types: t.Tuple[st.Type, float],
name: t.Optional[str] = None,
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Integer(
self,
min: int,
max: int,
base: st.IntegerBase,
possible_values: t.Iterable[int],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
def Null(
self, properties: t.Optional[t.Mapping[str, str]] = None
) -> None:
pass
def Float(
self,
min: float,
max: float,
base: st.FloatBase,
possible_values: t.Iterable[float],
properties: t.Optional[t.Mapping[str, str]] = None,
) -> None:
pass
visitor = AddFK()
_type.accept(visitor)
return visitor.result