Skip to content
Snippets Groups Projects
Select Git revision
  • 12a2097b410139eae70dbd031a916b9d3a4487ff
  • main default protected
  • experiment-step-sharing
  • experiment-color-coded-graph
  • experiment-separated-graph
5 results

node_queries.py

Blame
    • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    node_queries.py 3.70 KiB
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80 
    

    

    

    
from neo4j import Driver
from neo4j_queries.utils import clean_component_id

def ensure_component_node(driver: Driver, prefixed_component_id: str) -> tuple[int,str]:
    """
    Ensures that there exists a component node corresponding to the file with local path prefixed_component_id.
    The ID of the component can be given based on the local relative path, so it is cleaned 
    before querying Neo4j.

    Parameters:
        driver (Driver): the Neo4j driver
        prefixed_component_id (str): the local relative path of the component

    Returns:
        tuple[int,str]: the Neoj4 internal ID of the component node, the component ID of the component
    """
    component_id = clean_component_id(prefixed_component_id)
    query = """
    MERGE (c:Component {component_id: $component_id})
    RETURN elementId(c) AS node_internal_id, c.component_id AS component_id
    """
    with driver.session() as session:
        result = session.run(query, component_id=component_id)
        record = result.single()
        return record["node_internal_id"], record["component_id"]

def ensure_parameter_node(driver: Driver, node_id: str, prefixed_component_id: str, param_type: str) \
        -> tuple[int,str,str,str]: 
    """
    Ensures that there exists a parameter node with ID node_id and type param_type
    associated with the component in the file with local path prefixed_component_id.
    The ID of the component can be given based on the local relative path, so it is cleaned 
    before querying Neo4j.

    Parameters:
        driver (Driver): the Neo4j driver
        node_id (str): the ID of the parameter
        prefixed_component_id (str): the local relative path of the component
        param_type (str): the type of the parameter ('in' or 'out')

    Returns:
        tuple[int,str,str, str]: the Neoj4 internal ID of the parameter node, the parameter ID, the component ID, the parameter type
    """
    component_id = clean_component_id(prefixed_component_id)
    query = """
    MERGE (n:Parameter {parameter_id: $node_id, component_id: $component_id, parameter_type: $param_type})
    RETURN elementId(n) AS node_internal_id, n.parameter_id AS id_property, n.component_id AS component_id_property,
        n.parameter_type AS parameter_type_property
    """
    with driver.session() as session:
        result = session.run(query, node_id=node_id, component_id=component_id, param_type=param_type)
        record = result.single()
        return record["node_internal_id"], record["id_property"], record["component_id_property"], record['parameter_type_property']
    
def ensure_data_node(driver: Driver, node_id: str, prefixed_component_id: str) -> tuple[int,str,str]:
    """
    Ensures that there exists a data node with ID node_id
    associated with the component in the file with local path prefixed_component_id.
    The ID of the component can be given based on the local relative path, so it is cleaned 
    before querying Neo4j.

    Parameters:
        driver (Driver): the Neo4j driver
        node_id (str): the ID of the data 
        prefixed_component_id (str): the local relative path of the component

    Returns:
        tuple[int,str,str, str]: the Neoj4 internal ID of the data node, the data ID, the component ID
    """
    component_id = clean_component_id(prefixed_component_id)
    query = """
    MERGE (n:Data {data_id: $node_id, component_id: $component_id})
    RETURN elementId(n) AS node_internal_id, n.data_id AS id_property, n.component_id AS component_id_property
    """
    with driver.session() as session:
        result = session.run(query, node_id=node_id, component_id=component_id)
        record = result.single()
        return record["node_internal_id"], record["id_property"], record["component_id_property"]