Coverage for railway / cli / new.py: 77%

1"""railway new command implementation.""" 

2 

3from enum import Enum 

4from pathlib import Path 

5 

6import typer 

7 

8 

9class ComponentType(str, Enum): 

10 """Type of component to create.""" 

11 

12 entry = "entry" 

13 node = "node" 

14 

15 

16def _is_railway_project() -> bool: 

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

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

19 

20 

21def _write_file(path: Path, content: str) -> None: 

22 """Write content to a file.""" 

23 path.write_text(content) 

24 

25 

26def _get_entry_template(name: str) -> str: 

27 """Get basic entry point template.""" 

28 return f'''"""{name} entry point.""" 

29 

30from railway import entry_point, node 

31from loguru import logger 

32 

33 

34@entry_point 

35def main(): 

36 """ 

37 {name} entry point. 

38 

39 TODO: Add your implementation here. 

40 """ 

41 logger.info("Starting {name}") 

42 # Your implementation here 

43 return "Success" 

44 

45 

46if __name__ == "__main__": 

47 main() 

48''' 

49 

50 

51def _get_entry_example_template(name: str) -> str: 

52 """Get example entry point template.""" 

53 return f'''"""{name} entry point with example implementation.""" 

54 

55from datetime import datetime 

56 

57from railway import entry_point, node, pipeline 

58from loguru import logger 

59 

60 

61@node 

62def fetch_data(date: str) -> dict: 

63 """Fetch data for the given date.""" 

64 logger.info(f"Fetching data for {{date}}") 

65 # Example: Replace with actual API call 

66 return {{"date": date, "records": [1, 2, 3]}} 

67 

68 

69@node 

70def process_data(data: dict) -> dict: 

71 """Process the fetched data.""" 

72 logger.info(f"Processing {{len(data['records'])}} records") 

73 return {{ 

74 "date": data["date"], 

75 "summary": {{ 

76 "total": len(data["records"]), 

77 "sum": sum(data["records"]), 

78 }} 

79 }} 

80 

81 

82@entry_point 

83def main(date: str = None, dry_run: bool = False): 

84 """ 

85 {name} entry point. 

86 

87 Args: 

88 date: Target date (YYYY-MM-DD), defaults to today 

89 dry_run: If True, don't make actual changes 

90 """ 

91 if date is None: 

92 date = datetime.now().strftime("%Y-%m-%d") 

93 

94 if dry_run: 

95 logger.warning("DRY RUN mode - no actual changes") 

96 

97 result = pipeline( 

98 fetch_data(date), 

99 process_data, 

100 ) 

101 

102 logger.info(f"Result: {{result}}") 

103 return result 

104 

105 

106if __name__ == "__main__": 

107 main() 

108''' 

109 

110 

111def _get_node_template(name: str) -> str: 

112 """Get basic node template.""" 

113 return f'''"""{name} node.""" 

114 

115from railway import node 

116from loguru import logger 

117 

118 

119@node 

120def {name}(data: dict) -> dict: 

121 """ 

122 {name} node. 

123 

124 Args: 

125 data: Input data 

126 

127 Returns: 

128 Processed data 

129 

130 TODO: Add your implementation here. 

131 """ 

132 logger.info(f"Processing in {name}") 

133 # Your implementation here 

134 return data 

135''' 

136 

137 

138def _get_node_example_template(name: str) -> str: 

139 """Get example node template.""" 

140 return f'''"""{name} node with example implementation.""" 

141 

142from railway import node 

143from loguru import logger 

144 

145 

146@node(retry=True) 

147def {name}(data: dict) -> dict: 

148 """ 

149 {name} node. 

150 

151 This is an example node that demonstrates: 

152 - Type annotations 

153 - Logging 

154 - Error handling (via @node decorator) 

155 - Return value 

156 

157 Args: 

158 data: Input data dictionary 

159 

160 Returns: 

161 Processed data dictionary 

162 """ 

163 logger.info(f"Starting {name} with {{len(data)}} fields") 

164 

165 # Example processing 

166 result = {{ 

167 **data, 

168 "processed_by": "{name}", 

169 "status": "completed", 

170 }} 

171 

172 logger.debug(f"Processed result: {{result}}") 

173 return result 

174''' 

175 

176 

177def _get_node_test_template(name: str) -> str: 

178 """Get test template for a node.""" 

179 class_name = "".join(word.title() for word in name.split("_")) 

180 return f'''"""Tests for {name} node.""" 

181 

182import pytest 

183from unittest.mock import MagicMock, patch 

184 

185from src.nodes.{name} import {name} 

186 

187 

188class Test{class_name}: 

189 """Test suite for {name} node.""" 

190 

191 def test_{name}_success(self): 

192 """Test normal case.""" 

193 # TODO: Implement test 

194 pass 

195 

196 def test_{name}_error(self): 

197 """Test error case.""" 

198 # TODO: Implement test 

199 pass 

200''' 

201 

202 

203def _create_entry(name: str, example: bool, force: bool) -> None: 

204 """Create a new entry point.""" 

205 file_path = Path.cwd() / "src" / f"{name}.py" 

206 

207 if file_path.exists() and not force: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true

208 typer.echo(f"Error: {file_path} already exists. Use --force to overwrite.", err=True) 

209 raise typer.Exit(1) 

210 

211 content = _get_entry_example_template(name) if example else _get_entry_template(name) 

212 _write_file(file_path, content) 

213 

214 typer.echo(f"Created entry point: src/{name}.py") 

215 typer.echo("Entry point is ready to use\n") 

216 typer.echo("To run:") 

217 typer.echo(f" railway run {name}") 

218 typer.echo(f" # or: uv run python -m src.{name}") 

219 

220 

221def _create_node_test(name: str) -> None: 

222 """Create test file for node.""" 

223 tests_dir = Path.cwd() / "tests" / "nodes" 

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

225 tests_dir.mkdir(parents=True) 

226 

227 test_file = tests_dir / f"test_{name}.py" 

228 if test_file.exists(): 228 ↛ 229line 228 didn't jump to line 229 because the condition on line 228 was never true

229 return # Don't overwrite existing tests 

230 

231 content = _get_node_test_template(name) 

232 _write_file(test_file, content) 

233 

234 

235def _create_node(name: str, example: bool, force: bool) -> None: 

236 """Create a new node.""" 

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

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

239 nodes_dir.mkdir(parents=True) 

240 (nodes_dir / "__init__.py").write_text('"""Node modules."""\n') 

241 

242 file_path = nodes_dir / f"{name}.py" 

243 

244 if file_path.exists() and not force: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 typer.echo(f"Error: {file_path} already exists. Use --force to overwrite.", err=True) 

246 raise typer.Exit(1) 

247 

248 content = _get_node_example_template(name) if example else _get_node_template(name) 

249 _write_file(file_path, content) 

250 

251 # Create test file 

252 _create_node_test(name) 

253 

254 typer.echo(f"Created node: src/nodes/{name}.py") 

255 typer.echo(f"Created test: tests/nodes/test_{name}.py\n") 

256 typer.echo("To use in an entry point:") 

257 typer.echo(f" from src.nodes.{name} import {name}") 

258 

259 

260def new( 

261 component_type: ComponentType = typer.Argument(..., help="Type: entry or node"), 

262 name: str = typer.Argument(..., help="Name of the component"), 

263 example: bool = typer.Option(False, "--example", help="Generate with example code"), 

264 force: bool = typer.Option(False, "--force", help="Overwrite if exists"), 

265) -> None: 

266 """ 

267 Create a new entry point or node. 

268 

269 Examples: 

270 railway new entry daily_report 

271 railway new node fetch_data --example 

272 """ 

273 # Validate we're in a project 

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

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

276 raise typer.Exit(1) 

277 

278 if component_type == ComponentType.entry: 

279 _create_entry(name, example, force) 

280 else: 

281 _create_node(name, example, force)