Module pyracmon.graph

def new_graph(template: GraphTemplate, *bases: Union[Graph, GraphView]) -> Graph:
    """
    Create a graph from a template.

    Use this function instead of invoking constructor directly.

    Args:
        template: A template of a graph.
        bases: Other graphs whose nodes are appended to created graph.
    Returns:
        Created graph.
    """
    graph = Graph(template)

    for b in bases:
        graph += b

    return graph

class Graph:
    """
    This class represents a graph composed of tree-structured node containers.

    The structure is determined by `GraphTemplate`. Use `new_graph` Instead of constructor to create new graph instance.

    ```python
    template = GraphSpac().new_template(
        a = (int, lambda x:x),
        b = (str, lambda x:x),
        c = (str, lambda x:x),
    )
    template.a << template.b << template.c
    graph = new_graph(template)
    ```

    In above code, a graph which has 3 properties ( `a` `b` `c` ) and a structure where `a` is parent of `b` and `b` is parent of `c` is created.

    `append` ( `replace` ) is a method to store entities in the graph with tying them each other according to the structure.
    Entites are encapsulated by `Node` which can have an edge to parent node.

    ```python
    graph.append(a=1, b="a", c="x").append(a=2, b="b", c="y")
    ```

    In `append`, entities are first sorted in descending order, and then:

    - Search a node whose entity is *identical* to the first entity from the corresponding node container.
        - If found, new node is not created and the *identical* node is set to next parent.
        - Otherwise, new node is appended and it is set to next parent.
    - Apply this to following entities in order. A difference is that *identical* node is searched from the sequence of parents in the session.

    In example here, the identification is done by entity value itself ( `lambda x:x` ). Next code is the example where *identical* nodes are found.

    ```python
    graph.append(a=1, b="a", c="z").append(a=2, b="c", c="y")
    ```

    In the first `append`, `a` and `b` has its *identical* node and `a` is *identical* in the second.
    `c` in the second one is not *identical* to any node because parent node `b="c"` is already added as new node.

    Due to the identification mechanism, entity relationships in the graph is guaranteed after repeating `append` .
    """
    def __init__(self, template: GraphTemplate):
        #: Graph template.
        self.template: GraphTemplate = template
        #: A `dict` containing node containers by their names.
        self.containers: dict[str, NodeContainer] = {p.name:self._to_container(p) for p in template}
        self._view = None

    def _to_container(self, prop: GraphTemplate.Property) -> 'NodeContainer':
        if isinstance(prop.kind, GraphTemplate):
            return _GraphNodeContainer(prop)
        else:
            return NodeContainer(prop)

    def _container_of(self, prop: GraphTemplate.Property) -> Optional['NodeContainer']:
        candidates = [c for c in self.containers.values() if c.prop.is_compatible(prop)]
        if len(candidates) > 1:
            raise ValueError(f"Container can't be determined from property '{prop.name}'.")
        return candidates[0] if candidates else None

    def __add__(self, another: Union[Self, GraphView]) -> 'Graph':
        """
        Create new graph by adding this graph and another graph.

        New graph has the same template as this graph's.
        On the other hand, because this method depends on `__iadd__()`, another graph must not have the same template.

        Args:
            another: Graph or its view.
        Returns:
            Created graph.
        """
        graph = Graph(self.template)

        graph += self
        graph += another

        return graph

    def __iadd__(self, another: Union[Self, GraphView]) -> Self:
        """
        Append nodes from another graph.

        Nodes of another graph are traversed from its root and appended to compatible containers each other.

        Args:
            another: Graph or its view.
        Returns:
            This graph.
        """
        graph = another if isinstance(another, Graph) else another()

        def add(n: Node, anc: dict[str, list[Node]]):
            c = self._container_of(n.prop)
            if c:
                c.append(n.entity, anc)
            for ch_ in n.children.values():
                for m in ch_.nodes:
                    add(m, anc.copy())

        for c_ in graph.roots:
            for n_ in c_.nodes:
                add(n_, {})

        return self

    @property
    def roots(self) -> Iterable['NodeContainer']:
        """
        Returns root node containers.
        """
        return filter(lambda c: c.prop.parent is None, self.containers.values())

    @property
    def view(self) -> GraphView:
        """
        Returns an unmodifiable view of this graph.

        The view object works as the accessor to graph nodes.

        ```python
        >>> template = GraphSpac().new_template(a=int, b=str, c=str)
        >>> template.a << template.b
        >>> graph = new_graph(template)
        >>> view = graph.view
        >>> assert view() is graph                        # invocation
        >>> assert view.a is graph.containers["a"].view   # attribute
        >>> assert [c().name for c in view] == ["a", "c"] # iteration
        ```
        """
        if self._view is None:
            graph = self
            class _GraphView:
                def __call__(self) -> Graph:
                    """Returns the greph of this view."""
                    return graph
                def __iter__(self) -> Iterator[tuple[str, ContainerView[NodeContainer]]]:
                    """Iterates views of root containers."""
                    return map(lambda c: (c.name, c.view), filter(lambda c: c.prop.parent is None, graph.containers.values()))
                def __getattr__(self, name: str) -> ContainerView:
                    """Returns a view of a container of the name."""
                    return graph.containers[name].view
            self._view = _GraphView()
        return self._view

    def _append(self, to_replace: bool, entities: dict[str, Any]) -> Self:
        props = [p for p in self.template if p.name in entities]

        filtered = set()
        for p in props:
            if (p.parent is None) or (p.parent.name not in entities) or (p.parent.name in filtered):
                if p.entity_filter is None or p.entity_filter(entities[p.name]):
                    filtered.add(p.name)

        ancestors = {}
        for k in [p.name for p in props if p.name in filtered]:
            self.containers[k].append(entities[k], ancestors, to_replace)

        return self

    def append(self, **entities: Any) -> Self:
        """
        Append entities with associated property names.

        Args:
            entities: Entities keyed with associated property names.
        Returns:
            This graph.
        """
        return self._append(False, entities)

    def replace(self, **entities: Any) -> Self:
        """
        Works similarly to `append`, but entities of identical nodes are replaced with given entities.

        Args:
            entities: Entities keyed with associated property names.
        Returns:
            This graph.
        """
        return self._append(True, entities)

class GraphSpec:
    """
    This class contains the specifications of graph which control various behaviors in the lifecycles of graphs.

    3 kinds of functions are the core of graph behaviors: *identifier*, *entity filter* and *serializer* .

    *Identifier* and *entity filter* are used when appending values into graph.
    *Identifier* is a function to get a value used for the identification of graph entity. See `Graph` to know how this works.
    *Entity fliter* is a function to determine whether the entity should be appended to a graph or not.
    If `False` is returned for an entity, it is just ignored.

    See `pyracmon.graph.serialize` to know the detail of *serializer*.

    Each of them is bound to a `type` on registration to this and it affects nodes whose property type conforms to the `type` .
    """
    def __init__(
        self,
        identifiers: Optional[list[tuple[type, Identifier]]] = None,
        entity_filters: Optional[list[tuple[type, EntityFilter]]] = None,
        serializers: Optional[list[tuple[type, Serializer]]] = None,
    ):
        #: A list of pairs of type and *identifier*.
        self.identifiers: list[tuple[type, Identifier]] = identifiers or []
        #: A list of pairs of type and *entity_filter*.
        self.entity_filters: list[tuple[type, EntityFilter]] = entity_filters or []
        #: A list of pairs of type and *serializer*.
        self.serializers: list[tuple[type, Serializer]] = serializers or []

    def _get_inherited(self, holder: list[tuple[type, T]], t: type) -> Optional[T]:
        if not isinstance(t, type):
            return None
        return next(map(lambda x:x[1], filter(lambda x:issubtype(t, x[0]), holder)), None)

    def get_identifier(self, t: type) -> Optional[Callable[[Any], Any]]:
        """
        Returns the most appropriate identifier for a type.
        
        Args:
            t: Type of an entity.
        Returns:
            Identifier if exists.
        """
        return self._get_inherited(self.identifiers, t)

    def get_entity_filter(self, t: type) -> Optional[Callable[[Any], bool]]:
        """
        Returns the most appropriate entity filter for a type.
        
        Args:
            t: Type of an entity.
        Returns:
            Entity filter if exists.
        """
        return self._get_inherited(self.entity_filters, t)

    def find_serializers(self, t: type) -> list[Serializer]:
        """
        Returns a list of serializers applicable to a type.
        
        Args:
            t: Type of an entity.
        Returns:
            Serializers found.
        """
        if not isinstance(t, type):
            return []
        return list(map(lambda x:x[1], filter(lambda x:issubtype(t, x[0]), self.serializers[::-1])))

    def add_identifier(self, c: type, f: Callable[[Any], Any]) -> Self:
        """
        Register an identifier with a type.

        Args:
            c: A type bound to the identifier.
            f: An identifier function.
        Returns:
            This instance.
        """
        self.identifiers[0:0] = [(c, f)]
        return self

    def add_entity_filter(self, c: type, f: Callable[[Any], bool]) -> Self:
        """
        Register an entity filter with a type.

        Args:
            c: A type bound to the identifier.
            f: An entity filter function.
        Returns:
            This instance.
        """
        self.entity_filters[0:0] = [(c, f)]
        return self

    def add_serializer(self, c: type, f: Union[Serializer, NodeSerializer]) -> Self:
        """
        Register a serializer with a type.

        Args:
            c: A type bound to the identifier.
            f: A serializer function.
        Returns:
            This instance.
        """
        if isinstance(f, NodeSerializer):
            f = f.serializer
        self.serializers[0:0] = [(c, f)]
        return self

    def _make_policy(self, t: type, f: Union[IdentifyPolicy, Callable[[Any], Any], None]) -> IdentifyPolicy:
        f = f or self.get_identifier(t)

        if isinstance(f, IdentifyPolicy):
            return f
        elif callable(f):
            return HierarchicalPolicy(f)
        else:
            return neverPolicy()

    def _get_property_definition(self, definition: Union[
        TemplateProperty,
        type,
        GraphTemplate,
    ]) -> tuple[TypeDef, IdentifyPolicy, Optional[EntityFilter]]:
        if isinstance(definition, GraphTemplate):
            return definition, neverPolicy(), None
        elif isinstance(definition, type):
            return definition, self._make_policy(definition, None), self.get_entity_filter(definition)
        elif isinstance(definition, tuple):
            # python < 3.10
            if len(definition) == 3:
                kind, identifier, entity_filter = definition
            elif len(definition) == 2:
                kind, identifier, entity_filter = definition + (None,)
            elif len(definition) == 1:
                kind, identifier, entity_filter = definition + (None, None)
            elif len(definition) == 0:
                kind, identifier, entity_filter = (object, None, None)
            else:
                raise ValueError(f"Invalid value was found in keyword arguments of new_template().")
            # python >= 3.10
            #match definition:
            #    case (k, ident, ef):
            #        kind = k; identifier = ident; entity_filter = ef
            #    case (k, ident):
            #        kind = k; identifier = ident; entity_filter = None
            #    case (k,):
            #        kind = k; identifier = None; entity_filter = None
            #    case ():
            #        kind = object; identifier = None; entity_filter = None
            #    case _:
            #        raise ValueError(f"Invalid value was found in keyword arguments of new_template().")
            return kind, self._make_policy(kind, identifier), entity_filter or self.get_entity_filter(kind)
        else:
            raise ValueError(f"Invalid value was found in keyword arguments of new_template().")

    def new_template(self, *bases: GraphTemplate, **properties: Union[TemplateProperty, type, GraphTemplate]) -> GraphTemplate:
        """
        Creates a graph template with definitions of template properties.

        Each keyword argument corresponds to a template property where the key is proprety name and value is property definition.

        Property definition can be a `type` object or a tuple of at most 3 values.
        The former is the equivalent to a tuple which contains the `type` object alone.
        Values in the tuple are interpreted into following attributes in order.

        - The kind of property which indicates a type of entity.
        - *Identifier* of the property.
        - *Entity filter* of the property. 

        Omitted values are completed with registered items in this object.

        ```python
        template = GraphSpac().new_template(
            a = int,
            b = (str, lambda x:x),
            c = (str, lambda x:x, lambda x:len(x)>5),
        )
        ```

        Args:
            bases: Base templates whose properties and relations are merged into new template.
            properties: Definitions of template properties.
        Returns:
            Created graph template.
        """
        base = sum(bases, GraphTemplate([]))

        return base + GraphTemplate([(n, *self._get_property_definition(d)) for n, d in properties.items()])

    def to_dict(self, graph: GraphView, _params_: dict[str, dict[str, Any]] = {}, **settings: NodeSerializer) -> dict[str, Any]:
        """
        Serialize a graph into a `dict` .

        Only nodes whose names appear in keys of `settings` are serialized into the result.
        Each `NodeSerializer` object can be built by factory methods on `pyracmon.graph.serialize.S`.

        ```python
        GraphSpec().to_dict(
            graph,
            a = S.of(),
            b = S.name("B"),
        )
        ```

        Args:
            graph: A view of the graph.
            _params_: Parameters passed to `SerializationContext` and used by *serializer*s.
            settings: `NodeSerializer` for each property.
        Returns:
            Serialization result.
        """
        return SerializationContext(settings, self.find_serializers, _params_).execute(graph)

    def to_schema(self, template: GraphTemplate, **settings: NodeSerializer) -> GraphSchema:
        """
        Creates `GraphSchema` representing the structure of serialization result under given settings.

        Args:
            template: Template of a graph.
            settings: `NodeSerializer` for each property.
        Returns:
            Schema of serialization result.
        """
        return GraphSchema(self, template, **settings)

class GraphTemplate:
    """
    This class specifies the structure of a graph.

    The template is composed of template properties each of which corresponds to a node container of a graph.
    Each template property can be obtained via an attribute of its name from the template.

    Applying shift operator between properties creates the parent-child relationship between them.
    In next code, the template is composed of 4 properties where `d` is a child of `c`, and `b` and `c` are children of `a`.

    ```python
    template = GraphSpec().new_template(a=int, b=str, c=int, d=float)
    template.a << [template.b, template.c]
    template.c << template.d
    ```

    Templates are merged when `+` is applied to them. The result has properties defined in both templates with keeping their relationships.
    Merging of templates having properties of the same name fails by raising `ValueError`.

    Use `GraphSpec.new_template` or other factory functions to create a template instead of using constructor directly.
    """
    class Property:
        """
        Template property which determines various behaviors of graph nodes.
        """
        def __init__(
            self,
            template: 'GraphTemplate',
            name: str,
            kind: Union[type[T], 'GraphTemplate'],
            policy: IdentifyPolicy,
            entity_filter: Optional[Callable[[T], bool]],
            origin: Optional[Self] = None,
        ):
            #: Graph template this property belongs to.
            self.template = template
            #: Property name.
            self.name = name
            #: Graph node bound to this property should have entity of this type.
            self.kind = kind
            #: Policy of entity identification.
            self.policy = policy
            #: Entity filter function.
            self.entity_filter = entity_filter
            self._origin = origin

        def _assert_canbe_parent(self, another):
            if another.parent is not None:
                raise ValueError(f"Graph template property can not have multiple parents.")
            if self.template != another.template:
                raise ValueError(f"Properties can make parent-child relationship only when they are declared in the same template.")
            if self == another:
                raise ValueError(f"Recursive relationship is not allowed.")
            if isinstance(self.kind, GraphTemplate):
                raise ValueError(f"Property for graph template can't have child.")
            p = self
            while p.parent is not None:
                if p.parent == another:
                    raise ValueError(f"Recursive relationship is not allowed.")
                p = p.parent

        @property
        def parents(self) -> Iterable['GraphTemplate.Property']:
            """
            Returns all parent properties.
            """
            return map(lambda r: r[1], filter(lambda r: r[0] == self, self.template._relations))

        @property
        def parent(self) -> Optional['GraphTemplate.Property']:
            """
            Returns parent property if exists.
            """
            return next(iter(self.parents), None)

        @property
        def children(self) -> list['GraphTemplate.Property']:
            """
            Returns child properties.
            """
            return [r[0] for r in self.template._relations if r[1] == self]

        @property
        def origin(self) -> 'GraphTemplate.Property':
            """
            Returns the original property in the property chain generated by adding templates.
            """
            p = self
            while p._origin:
                p = p._origin
            return p

        def is_compatible(self, other):
            return self.origin is other.origin

        def _move_template(self, dest: 'GraphTemplate', new_name=None) -> 'GraphTemplate.Property':
            """
            Copy this property and children with setting their templates to `dest` .
            """
            new_name = new_name or self.name
            prop = GraphTemplate.Property(dest, new_name, self.kind, self.policy, self.entity_filter, origin=self)
            _set_template_property(dest, prop)
            for c in self.children:
                cc = c._move_template(dest)
                prop << cc # pyright: ignore [reportUnusedExpression]
            return prop

        @overload
        def __lshift__(self, children: 'GraphTemplate.Property') -> 'GraphTemplate.Property': ...
        @overload
        def __lshift__(self, children: list['GraphTemplate.Property']) -> list['GraphTemplate.Property']: ...
        def __lshift__(self, children: Union['GraphTemplate.Property', list['GraphTemplate.Property']]) -> Union['GraphTemplate.Property', list['GraphTemplate.Property']]:
            """
            Makes this property as a parent of given properties.

            Args:
                children: Property or properties to be children of this property.
            Returns:
                The same object as the argument.
            """
            targets = [children] if isinstance(children, GraphTemplate.Property) else children
            for c in targets:
                self._assert_canbe_parent(c)
            self.template._relations += [(c, self) for c in targets]
            return children

        def __rshift__(self, parent: 'GraphTemplate.Property') -> 'GraphTemplate.Property':
            """
            Makes this property as a child of another property.

            Args:
                parent: A Property to be a parent of this property.
            Returns:
                The same object as the argument.
            """
            parent._assert_canbe_parent(self)
            self.template._relations += [(self, parent)]
            return parent

        def __rrshift__(self, children: Union['GraphTemplate.Property', list['GraphTemplate.Property']]) -> 'GraphTemplate.Property':
            """
            Reversed version of `__lshift__()` prepared to locate a list of properties on the left side.

            Args:
                children: Property or properties to be children of this property.
            Returns:
                The same object as the argument.
            """
            self.__lshift__(children)
            return self

    def __init__(self, definitions: list[tuple[
        str,
        Union[type[T], Self, Property],
        Optional[IdentifyPolicy],
        Optional[Callable[[T], bool]],
    ]]):
        """
        Construct template with its properties.  Don't use this constructor directly.

        Args:
            definitions: Definitions of template properties.
        """
        self._properties: dict[str, GraphTemplate.Property] = {}
        self._relations: list[tuple[GraphTemplate.Property, GraphTemplate.Property]] = []

        for d in definitions:
            name, kind, ident, ef = d

            ident = ident or neverPolicy()

            if isinstance(kind, GraphTemplate):
                prop = GraphTemplate.Property(self, name, kind, ident, None)
                _set_template_property(self, prop)
            elif isinstance(kind, GraphTemplate.Property):
                kind._move_template(self, name)
            else:
                _set_template_property(self, GraphTemplate.Property(self, name, kind, ident, ef))

    def __getattr__(self, key) -> 'GraphTemplate.Property':
        return self._properties[key]

    def __iter__(self) -> Iterator['GraphTemplate.Property']:
        """
        Iterates properties in parent-to-child order.

        Returns:
            Property iterator.
        """
        return _walk_properties(self._properties)

    def __iadd__(self, another: 'GraphTemplate') -> Self:
        """
        Adds another template to this template.

        Args:
            another: Another template.
        Returns:
            This instance.
        """
        for p in another._properties.values():
            prop = GraphTemplate.Property(self, p.name, p.kind, p.policy, p.entity_filter, origin=p)
            _set_template_property(self, prop)

        for n, p in another._relations:
            getattr(self, n.name) >> getattr(self, p.name) # pyright: ignore [reportUnusedExpression]

        return self

    def __add__(self, another: 'GraphTemplate') -> 'GraphTemplate':
        """
        Creates new template by merging this template and another one.

        Args:
            another: Another template.
        Returns:
            New template.
        """
        template = GraphTemplate([])
        template += self
        template += another

        return template

class IdentifyPolicy:
    """
    Provides entity identification functionalities used during appending entities to a graph.

    Identification mechanism is based on the equality of identification keys extracted by entities.
    """
    def __init__(self, identifier: Optional[Callable[[Any], Any]]):
        #: A function to extract the identification key from an entity.
        self.identifier = identifier

    def get_identifier(self, value: Any) -> Any:
        """
        Returns identification key from an entity.

        Args:
            value: An entity.
        Returns:
            Identification key.
        """
        return self.identifier(value) if self.identifier else None

    def identify(
        self,
        prop: NodePropType,
        candidates: Iterable[MN],
        ancestors: Mapping[str, Iterable[MapNodeType[MapNodeType[MN, str], str]]],
    ) -> tuple[list[Optional[MN]], list[MN]]:
        """
        Select parent nodes and identical nodes of a new entity.

        This method is called during appending an entity to a graph.

        Args:
            prop: Template property for new entity.
            candidates: Nodes having the same identification key as the key of new entity.
            ancestors: Parent nodes mapped by property names.
        Returns
            The first item is a list of Parent nodes to which the node of new entity should be appended newly.
            `None` means to append a new node without parent. The second item is a list of identical nodes,
            which will be merged into ancestors and used in subsequent identifications of child entities.
        """
        raise NotImplementedError()

class S(metaclass=SerializerMeta):
    """
    An utility class to build `NodeSerializer` .

    This class provides factory class methods to create `NodeSerializer`
    each of which works in the same way as the method of the same name declared on `NodeSerializer` .

    Use them to supply `NodeSerializer`s to functions to serialize a graph or to create a graph schema
    such as `graph_dict` or `graph_schema` .

    ```python
    graph_dict(
        graph,
        a = S.of(),
        b = S.head(),
    )
    ```
    """
    @classmethod
    def of(
        cls,
        namer: Optional[Union[str, Callable[[str], str]]] = None,
        aggregator: Optional[Union[Callable[[list[Node]], Node], Callable[[list[Node]], list[Node]]]] = None,
        *serializers: Serializer,
    ) -> 'NodeSerializer':
        """
        Create an instance of `NodeSerializer`.

        Args:
            namer: A string or naming function.
            aggregator: An aggregation function or an index of node to select in node container.
            serializer: A list of *serializer* s.
        Returns:
            Created `NodeSerializer` .
        """
        return NodeSerializer(namer, aggregator, *serializers)