Coverage for railway / cli / list.py: 74%

1"""railway list command implementation.""" 

2 

3import ast 

4from pathlib import Path 

5from typing import Optional 

6 

7import typer 

8 

9 

10def _is_railway_project() -> bool: 

11 """Check if current directory is a Railway project.""" 

12 return (Path.cwd() / "src").exists() 

13 

14 

15def _extract_module_docstring(content: str) -> str | None: 

16 """Extract module docstring from Python code.""" 

17 try: 

18 tree = ast.parse(content) 

19 docstring = ast.get_docstring(tree) 

20 if docstring: 20 ↛ 23line 20 didn't jump to line 23 because the condition on line 20 was always true

21 # Return first line only 

22 return docstring.split("\n")[0].strip() 

23 return None 

24 except Exception: 

25 return None 

26 

27 

28def _analyze_entry_file(file_path: Path) -> dict | None: 

29 """Analyze a Python file for @entry_point decorator.""" 

30 try: 

31 content = file_path.read_text() 

32 

33 # Check for @entry_point decorator 

34 if "@entry_point" not in content: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 return None 

36 

37 # Get module docstring 

38 docstring = _extract_module_docstring(content) 

39 

40 return { 

41 "name": file_path.stem, 

42 "path": str(file_path.relative_to(Path.cwd())), 

43 "description": docstring or "No description", 

44 } 

45 except Exception: 

46 return None 

47 

48 

49def _analyze_node_file(file_path: Path) -> dict | None: 

50 """Analyze a Python file for @node decorator.""" 

51 try: 

52 content = file_path.read_text() 

53 

54 # Check for @node decorator 

55 if "@node" not in content: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 return None 

57 

58 # Get module docstring 

59 docstring = _extract_module_docstring(content) 

60 

61 return { 

62 "name": file_path.stem, 

63 "path": str(file_path.relative_to(Path.cwd())), 

64 "description": docstring or "No description", 

65 } 

66 except Exception: 

67 return None 

68 

69 

70def _find_entries() -> list[dict]: 

71 """Find all entry points in src/.""" 

72 src_dir = Path.cwd() / "src" 

73 skip_files = {"__init__.py", "settings.py"} 

74 

75 def should_analyze(py_file: Path) -> bool: 

76 return not py_file.name.startswith("_") and py_file.name not in skip_files 

77 

78 files = [f for f in src_dir.glob("*.py") if should_analyze(f)] 

79 entries = [_analyze_entry_file(f) for f in files] 

80 return [e for e in entries if e is not None] 

81 

82 

83def _find_nodes() -> list[dict]: 

84 """Find all nodes in src/nodes/.""" 

85 nodes_dir = Path.cwd() / "src" / "nodes" 

86 

87 if not nodes_dir.exists(): 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 return [] 

89 

90 def should_analyze(py_file: Path) -> bool: 

91 return not py_file.name.startswith("_") 

92 

93 files = [f for f in nodes_dir.glob("*.py") if should_analyze(f)] 

94 nodes = [_analyze_node_file(f) for f in files] 

95 return [n for n in nodes if n is not None] 

96 

97 

98def _count_tests() -> int: 

99 """Count test files.""" 

100 tests_dir = Path.cwd() / "tests" 

101 if not tests_dir.exists(): 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 return 0 

103 

104 return sum(1 for _ in tests_dir.rglob("test_*.py")) 

105 

106 

107def _display_entries(entries: list[dict]) -> None: 

108 """Display entry points.""" 

109 typer.echo("\nEntry Points:") 

110 if not entries: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 typer.echo(" (none)") 

112 return 

113 

114 for entry in entries: 

115 typer.echo(f" * {entry['name']:20} {entry['description']}") 

116 

117 

118def _display_nodes(nodes: list[dict]) -> None: 

119 """Display nodes.""" 

120 typer.echo("\nNodes:") 

121 if not nodes: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 typer.echo(" (none)") 

123 return 

124 

125 for node in nodes: 

126 typer.echo(f" * {node['name']:20} {node['description']}") 

127 

128 

129def _display_all(entries: list[dict], nodes: list[dict], tests: int) -> None: 

130 """Display all components.""" 

131 _display_entries(entries) 

132 _display_nodes(nodes) 

133 

134 typer.echo(f"\nStatistics:") 

135 typer.echo(f" {len(entries)} entry points, {len(nodes)} nodes, {tests} tests") 

136 

137 

138def list_components( 

139 filter_type: Optional[str] = typer.Argument(None, help="Filter: entries or nodes"), 

140) -> None: 

141 """ 

142 List entry points and nodes in the project. 

143 

144 Examples: 

145 railway list # Show all 

146 railway list entries # Show only entry points 

147 railway list nodes # Show only nodes 

148 """ 

149 if not _is_railway_project(): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true

150 typer.echo("Error: Not in a Railway project (src/ directory not found)", err=True) 

151 raise typer.Exit(1) 

152 

153 entries = _find_entries() 

154 nodes = _find_nodes() 

155 tests = _count_tests() 

156 

157 if filter_type == "entries": 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 _display_entries(entries) 

159 elif filter_type == "nodes": 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 _display_nodes(nodes) 

161 else: 

162 _display_all(entries, nodes, tests)