Coverage for railway / cli / run.py: 9%

1"""railway run command implementation.""" 

2 

3import importlib.util 

4import sys 

5from pathlib import Path 

6from typing import List, Optional 

7 

8import typer 

9 

10 

11def _find_project_root() -> Optional[Path]: 

12 """Find project root by looking for src/ directory.""" 

13 current = Path.cwd() 

14 

15 # Check current directory 

16 if (current / "src").exists(): 

17 return current 

18 

19 # Check parent directories 

20 for parent in current.parents: 

21 if (parent / "src").exists(): 

22 return parent 

23 # Stop at markers 

24 if (parent / "pyproject.toml").exists(): 

25 if (parent / "src").exists(): 

26 return parent 

27 break 

28 

29 return None 

30 

31 

32def _list_entries(project_root: Path) -> List[str]: 

33 """List available entries.""" 

34 src_dir = project_root / "src" 

35 entries = [] 

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

37 

38 for py_file in src_dir.glob("*.py"): 

39 if py_file.name.startswith("_"): 

40 continue 

41 if py_file.name in skip_files: 

42 continue 

43 entries.append(py_file.stem) 

44 

45 return entries 

46 

47 

48def _execute_entry(project_root: Path, entry_name: str, extra_args: List[str]) -> None: 

49 """Execute the entry point.""" 

50 # Add project root to sys.path 

51 src_path = str(project_root) 

52 if src_path not in sys.path: 

53 sys.path.insert(0, src_path) 

54 

55 # Load the module 

56 entry_path = project_root / "src" / f"{entry_name}.py" 

57 spec = importlib.util.spec_from_file_location(f"src.{entry_name}", entry_path) 

58 if spec is None or spec.loader is None: 

59 raise RuntimeError(f"Could not load module: {entry_path}") 

60 

61 module = importlib.util.module_from_spec(spec) 

62 sys.modules[f"src.{entry_name}"] = module 

63 

64 # Set sys.argv for the entry point 

65 original_argv = sys.argv 

66 sys.argv = [str(entry_path)] + list(extra_args or []) 

67 

68 try: 

69 spec.loader.exec_module(module) 

70 

71 # If module has main function, it may have been called via __name__ == "__main__" 

72 # If not, try to call it explicitly 

73 if hasattr(module, "main") and callable(module.main): 

74 # Check if main was already called 

75 pass # Usually __main__ block handles this 

76 finally: 

77 sys.argv = original_argv 

78 

79 

80def run( 

81 entry_name: str = typer.Argument(..., help="Name of the entry point to run"), 

82 project: Optional[str] = typer.Option( 

83 None, "--project", "-p", 

84 help="Path to the project root" 

85 ), 

86 extra_args: Optional[List[str]] = typer.Argument(None, help="Arguments to pass to entry"), 

87) -> None: 

88 """ 

89 Run an entry point. 

90 

91 This is a simpler alternative to 'uv run python -m src.entry_name'. 

92 

93 Examples: 

94 railway run daily_report 

95 railway run daily_report -- --date 2024-01-01 

96 railway run --project /path/to/project my_entry 

97 """ 

98 # Determine project root 

99 if project: 

100 project_root = Path(project).resolve() 

101 else: 

102 project_root = _find_project_root() 

103 

104 if project_root is None: 

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

106 typer.echo("Use --project to specify the project root", err=True) 

107 raise typer.Exit(1) 

108 

109 # Check entry exists 

110 entry_path = project_root / "src" / f"{entry_name}.py" 

111 if not entry_path.exists(): 

112 typer.echo(f"Error: Entry point '{entry_name}' not found at {entry_path}", err=True) 

113 typer.echo("\nAvailable entries:", err=True) 

114 entries = _list_entries(project_root) 

115 for entry in entries: 

116 typer.echo(f" • {entry}", err=True) 

117 raise typer.Exit(1) 

118 

119 # Run the entry 

120 typer.echo(f"Running entry point: {entry_name}") 

121 

122 try: 

123 _execute_entry(project_root, entry_name, extra_args or []) 

124 except Exception as e: 

125 typer.echo(f"Error: Failed to run entry: {e}", err=True) 

126 raise typer.Exit(1)