Coverage for fastagency/cli/docker_cli.py: 100%

3 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-04-19 12:16 +0000

1import subprocess # nosec B404 1defghiabc

2from logging import getLogger 1defghiabc

3from typing import Annotated, Optional 1defghiabc

4 

5import typer 1defghiabc

6 

7from .logging import setup_logging 1defghiabc

8 

9docker_app = typer.Typer(rich_markup_mode="rich") 1defghiabc

10 

11 

12setup_logging() 1defghiabc

13logger = getLogger(__name__) 1defghiabc

14 

15 

16@docker_app.command( 1defghiabc

17 context_settings={"allow_extra_args": True, "ignore_unknown_options": True}, 1defghiabc

18 help="Build a Docker image for the FastAgency app", 1defghiabc

19) 

20def build( 1defghiabc

21 build_context: Annotated[ 1defghiabc

22 str, 1defghiabc

23 typer.Argument( 1defghiabc

24 ..., 1defghiabc

25 help="Docker build context", 1defghiabc

26 ), 

27 ] = ".", 1abc

28 *, 

29 file: Annotated[ 1defghiabc

30 str, 1defghiabc

31 typer.Option( 1defghiabc

32 "--file", 1defghiabc

33 "-f", 1defghiabc

34 help="Name of the Dockerfile", 1defghiabc

35 ), 

36 ] = "docker/Dockerfile", 1defghiabc

37 tag: Annotated[ 1defghiabc

38 str, 1defghiabc

39 typer.Option( 1defghiabc

40 "--tag", 1defghiabc

41 "-t", 1defghiabc

42 help='Name and optionally a tag (format: "name:tag")', 1defghiabc

43 ), 

44 ] = "deploy_fastagency", 1defghiabc

45 progress: Annotated[ 1defghiabc

46 str, 1defghiabc

47 typer.Option( 1defghiabc

48 "--progress", 1defghiabc

49 help="Set type of progress output (auto, plain, tty, rawjson).", 1defghiabc

50 ), 

51 ] = "plain", 1defghiabc

52 ctx: typer.Context, 1defghiabc

53) -> None: 1defghiabc

54 command = [ 1dabc

55 "docker", 1dabc

56 "build", 1dabc

57 "-t", 1dabc

58 tag, 1dabc

59 "-f", 1dabc

60 file, 1dabc

61 "--progress", 1dabc

62 progress, 1dabc

63 build_context, 1dabc

64 ] 

65 command += ctx.args 1dabc

66 typer.echo( 1dabc

67 f"Building FastAgency Docker image with the command: {' '.join(command)}" 1dabc

68 ) 

69 try: 1dabc

70 # Run the docker build command 

71 result = subprocess.run( # nosec B603 1dabc

72 command, check=True, capture_output=True, text=True 1dabc

73 ) 

74 typer.echo(result.stdout) 1dabc

75 typer.echo(f"Image '{tag}' built successfully!") 1dabc

76 except subprocess.CalledProcessError as e: 

77 typer.echo(f"Error: {e.stderr}", err=True) 

78 raise typer.Exit(code=1) from e 

79 

80 

81@docker_app.command( 1defghiabc

82 context_settings={"allow_extra_args": True, "ignore_unknown_options": True}, 1defghiabc

83 help="Run a Docker container for the FastAgency app", 1defghiabc

84) 

85def run( 1defghiabc

86 image: Annotated[ 1defghiabc

87 str, 1defghiabc

88 typer.Argument( 1defghiabc

89 ..., 1defghiabc

90 help="The Docker image to run", 1defghiabc

91 ), 

92 ] = "deploy_fastagency", 1abc

93 *, 

94 name: Annotated[ 1defghiabc

95 str, 1defghiabc

96 typer.Option( 1defghiabc

97 "--name", 1defghiabc

98 help="Assign a name to the container", 1defghiabc

99 ), 

100 ] = "deploy_fastagency", 1defghiabc

101 env: Annotated[ 1defghiabc

102 Optional[list[str]], 1defghiabc

103 typer.Option( 1defghiabc

104 "--env", 1defghiabc

105 "-e", 1defghiabc

106 help="Set environment variables", 1defghiabc

107 show_default=False, 1defghiabc

108 ), 

109 ] = None, 1defghiabc

110 publish: Annotated[ 1defghiabc

111 Optional[list[str]], 1defghiabc

112 typer.Option( 1defghiabc

113 "--publish", 1defghiabc

114 "-p", 1defghiabc

115 help="Publish a container's port(s) to the host", 1defghiabc

116 show_default=False, 1defghiabc

117 ), 

118 ] = None, 1defghiabc

119 remove: Annotated[ 1defghiabc

120 bool, 1defghiabc

121 typer.Option( 1defghiabc

122 "--rm", 1defghiabc

123 help="Automatically remove the container and its associated anonymous volumes when it exits", 1defghiabc

124 is_flag=True, 1defghiabc

125 ), 

126 ] = False, 1defghiabc

127 detach: Annotated[ 1defghiabc

128 bool, 1defghiabc

129 typer.Option( 1defghiabc

130 "--detach", 1defghiabc

131 "-d", 1defghiabc

132 help="Run container in background and print container ID", 1defghiabc

133 is_flag=True, 1defghiabc

134 ), 

135 ] = True, 1defghiabc

136 network: Annotated[ 1defghiabc

137 Optional[str], 1defghiabc

138 typer.Option( 1defghiabc

139 "--network", 1defghiabc

140 help="Connect a container to a network", 1defghiabc

141 show_default=False, 1defghiabc

142 ), 

143 ] = None, 1defghiabc

144 ctx: typer.Context, 1defghiabc

145) -> None: 1defghiabc

146 # Construct the docker run command using the provided options 

147 command = ["docker", "run", "--name", name] 1dabc

148 

149 if env: 1dabc

150 for env_var in env: 

151 command.extend(["--env", env_var]) 

152 

153 if publish: 1dabc

154 for port in publish: 

155 command.extend(["--publish", port]) 

156 if "8888:8888" not in publish: 

157 command.extend(["--publish", "8888:8888"]) 

158 else: 

159 command.extend(["--publish", "8888:8888"]) 1dabc

160 

161 if remove: 1dabc

162 command.append("--rm") 1dabc

163 

164 if detach: 1dabc

165 command.append("--detach") 1dabc

166 

167 if network: 1dabc

168 command.extend(["--network", network]) 1dabc

169 

170 command += ctx.args 1dabc

171 command.append(image) 1dabc

172 

173 try: 1dabc

174 typer.echo( 1dabc

175 f"Running FastAgency Docker image with the command: {' '.join(command)}" 1dabc

176 ) 

177 # Run the docker command 

178 result = subprocess.run( # nosec B603 1dabc

179 command, check=True, capture_output=True, text=True 1dabc

180 ) 

181 typer.echo(result.stdout) 1dabc

182 except subprocess.CalledProcessError as e: 

183 typer.echo(f"Error: {e.stderr}", err=True) 

184 raise typer.Exit(code=1) from e 

185 

186 

187@docker_app.command( 1defghiabc

188 context_settings={"allow_extra_args": False, "ignore_unknown_options": False}, 1defghiabc

189 help="Deploy the Docker container for the FastAgency app to Fly.io", 1defghiabc

190) 

191def deploy( 1defghiabc

192 config_file: Annotated[ 1defghiabc

193 str, 1defghiabc

194 typer.Argument( 1defghiabc

195 ..., 1defghiabc

196 help="The Fly.io configuration file", 1defghiabc

197 ), 

198 ] = "fly.toml", 1abc

199 *, 

200 openai_api_key: Annotated[ 1defghiabc

201 str, 1defghiabc

202 typer.Option( 1defghiabc

203 "--openai-api-key", 1defghiabc

204 help="OpenAI API key", 1defghiabc

205 envvar="OPENAI_API_KEY", 1defghiabc

206 show_default=False, 1defghiabc

207 ), 

208 ], 

209 # ctx: typer.Context, 

210) -> None: 1defghiabc

211 launch_command = [ 1dabc

212 "fly", 1dabc

213 "launch", 1dabc

214 "--config", 1dabc

215 config_file, 1dabc

216 "--copy-config", 1dabc

217 "--yes", 1dabc

218 ] 

219 # launch_command += ctx.args 

220 

221 set_secret_command = ["fly", "secrets", "set", "OPENAI_API_KEY=" + openai_api_key] 1dabc

222 try: 1dabc

223 typer.echo( 1dabc

224 f"Deploying FastAgency Docker image to Fly.io with the command: {' '.join(launch_command)}" 1dabc

225 ) 

226 # Run the fly deploy command 

227 deploy_result = subprocess.run( # nosec B603 1dabc

228 launch_command, check=True, capture_output=True, text=True 1dabc

229 ) 

230 typer.echo(deploy_result.stdout) 1dabc

231 

232 typer.echo( 1dabc

233 f"Setting OpenAI API key with the command: {' '.join(set_secret_command)}" 1dabc

234 ) 

235 # Run the fly secrets set command 

236 set_secret_result = subprocess.run( # nosec B603 1dabc

237 set_secret_command, check=True, capture_output=True, text=True 1dabc

238 ) 

239 typer.echo(set_secret_result.stdout) 1dabc

240 typer.echo("Deployed FastAgency Docker image to Fly.io successfully!") 1dabc

241 except subprocess.CalledProcessError as e: 

242 typer.echo(f"Error: {e.stderr}", err=True) 

243 raise typer.Exit(code=1) from e