Coverage for flowr / cli / session_cmd.py: 100%

1"""Session subcommand group: init, show, set-state. 

2 

3Parses CLI args, dispatches to domain/infrastructure, formats output. 

4""" 

5 

6import argparse 

7import sys 

8from pathlib import Path 

9from typing import NoReturn 

10 

11from flowr.cli.output import format_json, format_text 

12from flowr.cli.resolution import DefaultFlowNameResolver, FlowNameNotFoundError 

13from flowr.domain.loader import load_flow_from_file, resolve_subflows 

14from flowr.domain.session import SessionStackFrame 

15from flowr.infrastructure.config import FlowrConfig 

16from flowr.infrastructure.session_store import ( 

17 SessionAlreadyExistsError, 

18 SessionNameNotFoundError, 

19 SessionNotFoundError, 

20 YamlSessionStore, 

21) 

22 

23 

24def add_session_parser(sub: argparse._SubParsersAction) -> None: 

25 """Add the session subcommand group to the argument parser.""" 

26 session_parser = sub.add_parser("session", help="Manage workflow sessions") 

27 session_sub = session_parser.add_subparsers(dest="session_command") 

28 

29 # session init 

30 p_init = session_sub.add_parser("init", help="Initialize a new session") 

31 p_init.add_argument("flow", help="Flow name or file path") 

32 p_init.add_argument( 

33 "--name", default=None, help="Session name or file path (default: from config)" 

34 ) 

35 

36 # session show 

37 p_show = session_sub.add_parser("show", help="Show current session state") 

38 p_show.add_argument( 

39 "--name", default=None, help="Session name or file path (default: from config)" 

40 ) 

41 p_show.add_argument( 

42 "--format", 

43 choices=["yaml", "json"], 

44 default="json", 

45 dest="output_format", 

46 help="Output format (default: json)", 

47 ) 

48 

49 # session set-state 

50 p_set = session_sub.add_parser("set-state", help="Update the current session state") 

51 p_set.add_argument("state", help="New state ID") 

52 p_set.add_argument( 

53 "--name", default=None, help="Session name or file path (default: from config)" 

54 ) 

55 

56 # session list 

57 p_list = session_sub.add_parser("list", help="List all sessions") 

58 p_list.add_argument( 

59 "--format", 

60 choices=["yaml", "json"], 

61 default="json", 

62 dest="output_format", 

63 help="Output format (default: json)", 

64 ) 

65 

66 

67def _error(msg: str) -> NoReturn: 

68 """Print error to stderr and exit with code 1.""" 

69 print(f"error: {msg}", file=sys.stderr) # noqa: T201 

70 sys.exit(1) 

71 

72 

73def cmd_session_init( 

74 args: argparse.Namespace, config: FlowrConfig, resolver: DefaultFlowNameResolver 

75) -> int: 

76 """Run session init subcommand. 

77 

78 Returns: 

79 Exit code: 0 on success, 1 on error. 

80 """ 

81 flows_dir = config.flows_path() 

82 sessions_dir = config.sessions_path() 

83 name = args.name or config.default_session 

84 

85 try: 

86 flow_path = resolver.resolve(args.flow, flows_dir) 

87 except FlowNameNotFoundError as exc: 

88 _error(str(exc)) 

89 

90 store = YamlSessionStore(sessions_dir) 

91 

92 try: 

93 session = store.init(flow_path, name) 

94 except SessionAlreadyExistsError as exc: 

95 _error(str(exc)) 

96 

97 flow = load_flow_from_file(flow_path) 

98 if flow.states and flow.states[0].flow is not None: 

99 all_flows = resolve_subflows(flow, flow_path) 

100 ref_stem = Path(flow.states[0].flow).stem 

101 subflow = next((f for f in all_flows if f.flow == ref_stem), None) 

102 if subflow is not None and subflow.states: 

103 frame = SessionStackFrame(flow=session.flow, state=session.state) 

104 initial_id = subflow.states[0].id 

105 session = session.push_stack(frame, initial_id, new_flow=subflow.flow) 

106 store.save(session) 

107 

108 output = { 

109 "flow": session.flow, 

110 "state": session.state, 

111 "name": session.name, 

112 "created_at": session.created_at, 

113 } 

114 print(format_text(output)) # noqa: T201 

115 return 0 

116 

117 

118def cmd_session_show( 

119 args: argparse.Namespace, config: FlowrConfig, _resolver: DefaultFlowNameResolver 

120) -> int: 

121 """Run session show subcommand. 

122 

123 Returns: 

124 Exit code: 0 on success, 1 on error. 

125 """ 

126 sessions_dir = config.sessions_path() 

127 name = args.name or config.default_session 

128 

129 store = YamlSessionStore(sessions_dir) 

130 try: 

131 session = store.load(name) 

132 except (SessionNotFoundError, SessionNameNotFoundError) as exc: 

133 _error(str(exc)) 

134 

135 output: dict = { 

136 "flow": session.flow, 

137 "state": session.state, 

138 "name": session.name, 

139 "stack": [{"flow": f.flow, "state": f.state} for f in session.stack], 

140 "created_at": session.created_at, 

141 "updated_at": session.updated_at, 

142 } 

143 

144 if args.output_format == "json": 

145 print(format_json(output)) # noqa: T201 

146 else: 

147 print(format_text(output)) # noqa: T201 

148 return 0 

149 

150 

151def cmd_session_set_state( 

152 args: argparse.Namespace, config: FlowrConfig, resolver: DefaultFlowNameResolver 

153) -> int: 

154 """Run session set-state subcommand. 

155 

156 Returns: 

157 Exit code: 0 on success, 1 on error. 

158 """ 

159 sessions_dir = config.sessions_path() 

160 name = args.name or config.default_session 

161 

162 store = YamlSessionStore(sessions_dir) 

163 try: 

164 session = store.load(name) 

165 except (SessionNotFoundError, SessionNameNotFoundError) as exc: 

166 _error(str(exc)) 

167 

168 # Validate that the requested state exists in the flow 

169 flows_dir = config.flows_path() 

170 try: 

171 flow_path = resolver.resolve(session.flow, flows_dir) 

172 except FlowNameNotFoundError as exc: 

173 _error(str(exc)) 

174 

175 flow = load_flow_from_file(flow_path) 

176 state_ids = {s.id for s in flow.states} 

177 if args.state not in state_ids: 

178 _error(f"State '{args.state}' not found in flow '{session.flow}'") 

179 

180 updated = session.with_state(args.state) 

181 store.save(updated) 

182 

183 output = { 

184 "flow": updated.flow, 

185 "state": updated.state, 

186 "updated_at": updated.updated_at, 

187 } 

188 print(format_text(output)) # noqa: T201 

189 return 0 

190 

191 

192def cmd_session_list( 

193 args: argparse.Namespace, config: FlowrConfig, _resolver: DefaultFlowNameResolver 

194) -> int: 

195 """Run session list subcommand. 

196 

197 Returns: 

198 Exit code: 0 on success, 1 on error. 

199 """ 

200 sessions_dir = config.sessions_path() 

201 

202 store = YamlSessionStore(sessions_dir) 

203 sessions = store.list_sessions() 

204 

205 rows = [ 

206 { 

207 "name": s.name, 

208 "flow": s.flow, 

209 "state": s.state, 

210 "updated_at": s.updated_at, 

211 } 

212 for s in sessions 

213 ] 

214 

215 if args.output_format == "json": 

216 print(format_json(rows)) # noqa: T201 

217 else: 

218 print(format_text(rows)) # noqa: T201 

219 return 0