add nix-flake-show utils
Signed-off-by: ahuston-0 <aliceghuston@gmail.com>
This commit is contained in:
		
							
								
								
									
										39
									
								
								app/common.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										39
									
								
								app/common.py
									
									
									
									
									
										Normal file
									
								
							@@ -0,0 +1,39 @@
 | 
				
			|||||||
 | 
					#!/usr/bin/env python3
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					"""common."""
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					import logging
 | 
				
			||||||
 | 
					import sys
 | 
				
			||||||
 | 
					from subprocess import PIPE, Popen
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def configure_logger(level: str = "INFO") -> None:
 | 
				
			||||||
 | 
					    """Configure the logger.
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        level (str, optional): The logging level. Defaults to "INFO".
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    logging.basicConfig(
 | 
				
			||||||
 | 
					        level=level,
 | 
				
			||||||
 | 
					        datefmt="%Y-%m-%dT%H:%M:%S%z",
 | 
				
			||||||
 | 
					        format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s",
 | 
				
			||||||
 | 
					        handlers=[logging.StreamHandler(sys.stdout)],
 | 
				
			||||||
 | 
					    )
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def bash_wrapper(command: str, path:str = '.') -> tuple[str, int]:
 | 
				
			||||||
 | 
					    """Execute a bash command and capture the output.
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        command (str): The bash command to be executed.
 | 
				
			||||||
 | 
					        path (str): The current working directory, '.' by default
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Returns:
 | 
				
			||||||
 | 
					        Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
 | 
				
			||||||
 | 
					        the error output (stderr) as a string (optional), and the return code as an integer.
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    # This is a acceptable risk
 | 
				
			||||||
 | 
					    process = Popen(command.split(), stdout=PIPE, stderr=PIPE,cwd = path)  # noqa: S603
 | 
				
			||||||
 | 
					    output, _ = process.communicate()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    return output.decode(), process.returncode
 | 
				
			||||||
							
								
								
									
										29
									
								
								app/flake_show.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										29
									
								
								app/flake_show.py
									
									
									
									
									
										Normal file
									
								
							@@ -0,0 +1,29 @@
 | 
				
			|||||||
 | 
					#!/usr/bin/env python3
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					import json
 | 
				
			||||||
 | 
					from app.common import bash_wrapper
 | 
				
			||||||
 | 
					import shutil
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def traverse_json_base(json_dict:dict ,path: list[str]):
 | 
				
			||||||
 | 
					    final_paths = []
 | 
				
			||||||
 | 
					    for (key,value) in json_dict.items():
 | 
				
			||||||
 | 
					        if isinstance(value,dict):
 | 
				
			||||||
 | 
					            keys = value.keys()
 | 
				
			||||||
 | 
					            if 'type' in keys and value['type'] in ['nixos-configuration','derivation']:
 | 
				
			||||||
 | 
					                final_paths += ['.'.join(path + [key])]
 | 
				
			||||||
 | 
					            else:
 | 
				
			||||||
 | 
					                final_paths += traverse_json_base(value,path+[key])
 | 
				
			||||||
 | 
					    return final_paths
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def traverse_json(json_dict: dict):
 | 
				
			||||||
 | 
					    return traverse_json_base(json_dict,[])
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def get_derivations(path_to_flake: str):
 | 
				
			||||||
 | 
					    nix_path = shutil.which("nix")
 | 
				
			||||||
 | 
					    flake_show = bash_wrapper(f"{nix_path} flake show --json", path = path_to_flake)
 | 
				
			||||||
 | 
					    if flake_show[1] != 0:
 | 
				
			||||||
 | 
					        raise RuntimeError("flake show returned non-zero exit code")
 | 
				
			||||||
 | 
					    flake_show_json = json.loads(flake_show[0])
 | 
				
			||||||
 | 
					    derivations = traverse_json(flake_show_json)
 | 
				
			||||||
 | 
					    return derivations
 | 
				
			||||||
		Reference in New Issue
	
	Block a user