Method upsert_edge

minirag/kg/gremlin_impl.py:394–431

Upsert an edge and its properties between two nodes identified by their names. Args: source_node_id (str): Name of the source node (used as identifier) target_node_id (str): Name of the target node (used as identifier) edge_data (dict): Dictionar

(
        self, source_node_id: str, target_node_id: str, edge_data: Dict[str, Any]
    )

        retry=retry_if_exception_type((GremlinServerError,)),
    )
    async def upsert_edge(
        self, source_node_id: str, target_node_id: str, edge_data: Dict[str, Any]
    ):
        """
        Upsert an edge and its properties between two nodes identified by their names.

        Args:
            source_node_id (str): Name of the source node (used as identifier)
            target_node_id (str): Name of the target node (used as identifier)
            edge_data (dict): Dictionary of properties to set on the edge
        """
        source_node_name = GremlinStorage._fix_name(source_node_id)
        target_node_name = GremlinStorage._fix_name(target_node_id)
        edge_properties = GremlinStorage._convert_properties(edge_data)

        query = f"""g
                 .V().has('graph', {self.graph_name})
                 .has('entity_name', {source_node_name}).as('source')
                 .V().has('graph', {self.graph_name})
                 .has('entity_name', {target_node_name}).as('target')
                 .coalesce(
                      __.select('source').outE('DIRECTED').where(__.inV().as('target')),
                      __.select('source').addE('DIRECTED').to(__.select('target'))
                  )
                  .property('graph', {self.graph_name})
                 {edge_properties}
                 """
        try:
            await self._query(query)
            logger.debug(
                "Upserted edge from {%s} to {%s} with properties: {%s}",
                source_node_name,
                target_node_name,
                edge_properties,
            )
        except Exception as e:
            logger.error("Error during edge upsert: {%s}", e)
            raise

    async def _node2vec_embed(self):
        print("Implemented but never called.")