Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
sarus_data_spec / sarus_data_spec / manager / ops / primary_keys.py
Size: Mime:
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 pk_visitor(
    _type: st.Type, curr_path: t.Optional[st.Path] = None
) -> t.List[st.Path]:
    """Go over the type and returns a list of primary keys"""

    class AddPK(st.TypeVisitor):
        result: t.List[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 unique and (reference is None) and (curr_path is not None):
                self.result = [curr_path]

        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.List[st.Path] = []
            if curr_path is None:
                for item_name, item_type in fields.items():
                    additional_paths = pk_visitor(
                        item_type,
                        curr_path=straight_path([item_name]),
                    )
                    result += additional_paths
            else:
                for item_name, item_type in fields.items():
                    additional_paths = pk_visitor(
                        item_type,
                        curr_path=straight_path(
                            [
                                *(
                                    element
                                    for element in curr_path.to_strings_list()[
                                        0
                                    ]
                                ),
                                item_name,
                            ]
                        ),
                    )
                    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.List[st.Path] = []
            if curr_path is None:
                for item_name, item_type in fields.items():
                    additional_paths = pk_visitor(
                        item_type,
                        curr_path=straight_path([item_name]),
                    )
                    result += additional_paths

            else:
                for item_name, item_type in fields.items():
                    additional_paths = pk_visitor(
                        item_type,
                        curr_path=straight_path(
                            [
                                *(
                                    element
                                    for element in curr_path.to_strings_list()[
                                        0
                                    ]
                                ),
                                item_name,
                            ]
                        ),
                    )
                    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 = pk_visitor(
                    type,
                    curr_path=straight_path([OPTIONAL_VALUE]),
                )
            else:
                additional_paths = pk_visitor(
                    type,
                    curr_path=straight_path(
                        [
                            *(
                                element
                                for element in curr_path.to_strings_list()[0]
                            ),
                            OPTIONAL_VALUE,
                        ]
                    ),
                )
            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 = pk_visitor(
                    type,
                    curr_path=straight_path([LIST_VALUES]),
                )
            else:
                additional_paths = pk_visitor(
                    type,
                    curr_path=straight_path(
                        [
                            *(
                                element
                                for element in curr_path.to_strings_list()[0]
                            ),
                            LIST_VALUES,
                        ]
                    ),
                )
            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 = pk_visitor(
                    type,
                    curr_path=straight_path([ARRAY_VALUES]),
                )
            else:
                additional_paths = pk_visitor(
                    type,
                    curr_path=straight_path(
                        [
                            *(
                                element
                                for element in curr_path.to_strings_list()[0]
                            ),
                            ARRAY_VALUES,
                        ]
                    ),
                )
            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 = AddPK()
    _type.accept(visitor)
    return visitor.result