Skip to content
Snippets Groups Projects
Commit ac0df547 authored by Chiara Liotta's avatar Chiara Liotta
Browse files

fix function headers

parent f1adb477
No related branches found
No related tags found
No related merge requests found
Showing
with 22 additions and 14 deletions
.venv/
repos/
**/__pycache__/
*.py[cod]
Neo4j-25ebc0db-Created-2024-11-17.txt
\ No newline at end of file
File deleted
File deleted
File deleted
File deleted
File deleted
from neo4j import Driver
from graph_creation.utils import create_input_nodes_and_relationships, process_source_relationship
from neo4j_queries.node_queries import ensure_component_node, ensure_data_node, ensure_parameter_node
from neo4j_queries.edge_queries import create_data_relationship, create_out_param_relationship
from pathlib import Path
def process_cwl_inputs(driver, cwl_entity: dict):
def process_cwl_inputs(driver: Driver, cwl_entity: dict) -> None:
component_id = cwl_entity['path']
if type(cwl_entity['inputs']) == list:
for input in cwl_entity['inputs']:
......@@ -13,7 +14,7 @@ def process_cwl_inputs(driver, cwl_entity: dict):
for key in cwl_entity['inputs'].keys():
create_input_nodes_and_relationships(driver, key, component_id)
def process_cwl_outputs(driver, cwl_entity: dict):
def process_cwl_outputs(driver: Driver, cwl_entity: dict) -> None:
component_id = cwl_entity['path']
for output in cwl_entity['outputs']:
if type(output) == dict:
......@@ -29,7 +30,7 @@ def process_cwl_outputs(driver, cwl_entity: dict):
for o in output['outputSource']:
process_source_relationship(driver, o, component_id, param_node_internal_id)
def process_cwl_steps(driver, cwl_entity: dict, repo: str):
def process_cwl_steps(driver: Driver, cwl_entity: dict, repo: str) -> None:
for step in cwl_entity['steps']:
combined_path = Path(repo) / step['run']
step_path = str(combined_path)
......
from neo4j import Driver
from graph_creation.cwl_parsing import get_cwl_from_repo
from graph_creation.cwl_processing import process_cwl_inputs, process_cwl_outputs, process_cwl_steps
from neo4j_queries.node_queries import ensure_component_node
def process_repos(repo_list: list, driver):
def process_repos(repo_list: list[str], driver: Driver) -> None:
cwl_entities = {}
for repo in repo_list:
cwl_entities[repo]= get_cwl_from_repo(repo)
......
from neo4j import Driver
from neo4j_queries.node_queries import ensure_data_node, ensure_parameter_node
from neo4j_queries.edge_queries import create_data_relationship, create_in_param_relationship
def create_input_nodes_and_relationships(driver, input_id, component_id):
def create_input_nodes_and_relationships(driver: Driver, input_id: str, component_id: str) -> None:
# Create in-parameter node i_node with id = i.id and component_id = c_node.id
param_node = ensure_parameter_node(driver, input_id, component_id, 'in')
param_node_internal_id = param_node[0]
......@@ -13,7 +14,7 @@ def create_input_nodes_and_relationships(driver, input_id, component_id):
# Create a data edge from i_data_node to i_node
create_data_relationship(driver, data_node_internal_id, param_node_internal_id)
def process_source_relationship(driver, source_id, component_id, param_node_internal_id):
def process_source_relationship(driver: Driver, source_id: str, component_id: str, param_node_internal_id: str) -> None:
data_node = ensure_data_node(driver, source_id, component_id)
data_node_internal_id = data_node[0]
create_data_relationship(driver, param_node_internal_id, data_node_internal_id)
\ No newline at end of file
......@@ -5,7 +5,7 @@ import os
import gitlab
import subprocess
def clone_repos(repo_list: list, folder_name: str):
def clone_repos(repo_list: list[str], folder_name: str):
gl = gitlab.Gitlab('https://git.astron.nl')
projects = gl.projects.list(iterator=True, get_all=True)
for project in projects:
......@@ -17,7 +17,7 @@ def clone_repos(repo_list: list, folder_name: str):
if __name__ == '__main__':
relevant_repos = ['ldv/imaging_compress_pipeline']
folder = 'repos'
clone_repos(relevant_repos)
clone_repos(relevant_repos, folder)
load_status = dotenv.load_dotenv("Neo4j-25ebc0db-Created-2024-11-17.txt")
if load_status is False:
......
File deleted
File deleted
File deleted
File deleted
from neo4j import Driver
from neo4j_queries.utils import clean_component_id
def create_in_param_relationship(driver, prefixed_component_id, parameter_internal_id):
def create_in_param_relationship(driver: Driver, prefixed_component_id: str, parameter_internal_id: int) -> tuple[str,str]:
component_id = clean_component_id(prefixed_component_id)
query = """
MATCH (c:Component {component_id: $component_id}), (p)
......@@ -14,7 +15,7 @@ def create_in_param_relationship(driver, prefixed_component_id, parameter_intern
record = result.single()
return record["component_id"], record["parameter_id"]
def create_out_param_relationship(driver, prefixed_component_id, parameter_internal_id):
def create_out_param_relationship(driver: Driver, prefixed_component_id: str, parameter_internal_id: int) -> tuple[str,str]:
component_id = clean_component_id(prefixed_component_id)
query = """
MATCH (c:Component {component_id: $component_id}), (p)
......@@ -28,7 +29,7 @@ def create_out_param_relationship(driver, prefixed_component_id, parameter_inter
record = result.single()
return record["component_id"], record["parameter_id"]
def create_data_relationship(driver, from_internal_node_id, to_internal_node_id):
def create_data_relationship(driver: Driver, from_internal_node_id: int, to_internal_node_id: int) -> tuple[int,int]:
query = """
MATCH (a), (b)
WHERE id(a) = $from_internal_node_id AND id(b) = $to_internal_node_id
......
from neo4j import Driver
from neo4j_queries.utils import clean_component_id
def ensure_component_node(driver, prefixed_component_id):
def ensure_component_node(driver: Driver, prefixed_component_id: str) -> tuple[int,str]:
component_id = clean_component_id(prefixed_component_id)
query = """
MERGE (c:Component {component_id: $component_id})
......@@ -12,7 +13,8 @@ def ensure_component_node(driver, prefixed_component_id):
record = result.single()
return record["node_internal_id"], record["id_property"]
def ensure_parameter_node(driver, node_id, prefixed_component_id, param_type):
def ensure_parameter_node(driver: Driver, node_id: str, prefixed_component_id: str, param_type: str) \
-> tuple[int,str,str,str]:
component_id = clean_component_id(prefixed_component_id)
query = """
MERGE (n:Parameter {parameter_id: $node_id, component_id: $component_id})
......@@ -27,7 +29,7 @@ def ensure_parameter_node(driver, node_id, prefixed_component_id, 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, node_id, prefixed_component_id):
def ensure_data_node(driver: Driver, node_id: str, prefixed_component_id: str) -> tuple[int,str,str]:
component_id = clean_component_id(prefixed_component_id)
query = """
MERGE (n:Data {data_id: $node_id, component_id: $component_id})
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment