Skip to content
Snippets Groups Projects
Select Git revision
  • 8f67bd7971f62cd90c1b1b4372d9a3e04718f98b
  • main default protected
  • L2SS-2357-expand-ruff
  • test-10rc4
  • L2SS-1730
  • L2SS-1650-allow-threads-during-io
  • play-with-merge-trains
  • v0.3.1
  • v0.3.0
  • v0.2.0
  • v0.1.0
11 results

interface.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    main.py 1.68 KiB
    from graph_creation.repo_processing import process_repos
    from neo4j import GraphDatabase
    import dotenv
    import os
    import gitlab
    import subprocess
    
    def clone_repos(repo_list: list[str], folder_name: str) -> None:
        """
        Given a list of relative paths to ASTRON GitLab repositories and the name of a folder,
        the mentioned repositories are cloned into the mentioned folder.
    
        Parameters:
        repo_list (list[str]): list of relative paths to ASTRON GitLab repositories
        folder_name (str): the name of the folder to clone the repos into
        """
        gl = gitlab.Gitlab('https://git.astron.nl')
        projects = gl.projects.list(iterator=True, get_all=True)
        for project in projects:
            repo_name = project.attributes['path_with_namespace']
            if repo_name in repo_list:
                git_url = project.ssh_url_to_repo
                subprocess.call(['git', 'clone', git_url, f'./{folder_name}/{repo_name}'])
    
    if __name__ == '__main__':
        relevant_repos = ['ldv/imaging_compress_pipeline']
        folder = 'repos'
        clone_repos(relevant_repos, folder)
    
        # Get the authentication details for Neo4j instance
        load_status = dotenv.load_dotenv("Neo4j-25ebc0db-Created-2024-11-17.txt")
        if load_status is False:
            raise RuntimeError('Environment variables not loaded.')
    
        URI = os.getenv("NEO4J_URI")
        AUTH = (os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD"))
    
        repo_paths = [f'{folder}/{path}' for path in relevant_repos]
        
        with GraphDatabase.driver(URI, auth=AUTH) as driver:
            driver.verify_connectivity()
            print("Connection established.")
            driver = GraphDatabase.driver(URI, auth=AUTH)
            process_repos(repo_paths, driver)
            driver.close()