1""" 

2Router for Deployment routes 

3""" 

4 

5import json 

6import shlex 

7import subprocess 

8from os import path, scandir 

9from shutil import rmtree 

10from typing import Any, Dict, Set 

11 

12from fastapi import APIRouter, Depends, status 

13from fastapi.encoders import jsonable_encoder 

14from fastapi.exceptions import HTTPException 

15from starlette.responses import Response 

16 

17from serverctl_deployd.config import Settings 

18from serverctl_deployd.dependencies import get_settings 

19from serverctl_deployd.models.deployments import (DBConfig, Deployment, 

20 UpdateDeployment) 

21from serverctl_deployd.models.exceptions import GenericError 

22 

23 

24def _merge_dicts( 

25 current: Dict[Any, Any], 

26 update: Dict[Any, Any] 

27) -> Dict[Any, Any]: 

28 """ 

29 Merges two dicts: update into current. 

30 Used for update_deployment() 

31 """ 

32 for key in update: 

33 if key in current: 

34 if isinstance(current[key], dict) and isinstance( 

35 update[key], dict): 

36 _merge_dicts(current[key], update[key]) 

37 elif update[key] and current[key] != update[key]: 

38 current[key] = update[key] 

39 return current 

40 

41 

42router: APIRouter = APIRouter( 

43 prefix="/deployments", 

44 tags=["deployments"] 

45) 

46 

47 

48@router.post( 

49 "/", 

50 responses={ 

51 status.HTTP_409_CONFLICT: {"model": GenericError} 

52 }, 

53 response_model=Deployment, 

54) 

55def create_deployment( 

56 deployment: Deployment, 

57 settings: Settings = Depends(get_settings) 

58) -> Deployment: 

59 """Create a deployment""" 

60 deployment_path = settings.deployments_dir.joinpath(deployment.name) 

61 try: 

62 deployment_path.mkdir(parents=True) 

63 except FileExistsError as dir_exists_error: 

64 raise HTTPException( 

65 status_code=status.HTTP_409_CONFLICT, 

66 detail="A deployment with same name already exists" 

67 ) from dir_exists_error 

68 

69 compose_path = deployment_path.joinpath("docker-compose.yml") 

70 compose_path.write_text(deployment.compose_file, 

71 encoding="utf-8") 

72 

73 if deployment.env_file: 

74 env_path = deployment_path.joinpath(".env") 

75 env_path.write_text(deployment.env_file, 

76 encoding="utf-8") 

77 

78 if deployment.databases: 

79 db_file = deployment_path.joinpath("databases.json") 

80 db_file.touch() 

81 

82 with open(db_file, "w", encoding="utf-8") as json_file: 

83 db_json = jsonable_encoder(deployment.databases) 

84 json.dump(db_json, json_file, indent=4) 

85 

86 return deployment 

87 

88 

89@router.get("/", response_model=Set[str]) 

90def get_deployments( 

91 settings: Settings = Depends(get_settings) 

92) -> Set[str]: 

93 """Get a list of all deployments""" 

94 deployments: Set[str] = set() 

95 if settings.deployments_dir.exists(): 

96 deployments.update( 

97 item.name for item in scandir(settings.deployments_dir) 

98 if item.is_dir()) 

99 return deployments 

100 

101 

102@router.get( 

103 "/{name}", 

104 responses={ 

105 status.HTTP_404_NOT_FOUND: {"model": GenericError} 

106 }, 

107 response_model=Dict[str, DBConfig] 

108) 

109def get_deployment( 

110 name: str, 

111 settings: Settings = Depends(get_settings) 

112) -> Dict[str, DBConfig]: 

113 """Get database details of a deployment""" 

114 deployment_path = settings.deployments_dir.joinpath(name) 

115 if not deployment_path.exists(): 

116 raise HTTPException( 

117 status_code=status.HTTP_404_NOT_FOUND, 

118 detail="Deployment does not exist" 

119 ) 

120 db_file = deployment_path.joinpath("databases.json") 

121 json_data: Dict[str, DBConfig] = {} 

122 if db_file.exists(): 

123 with open(db_file, "r", encoding="utf-8") as json_file: 

124 json_data = json.load(json_file) 

125 return json_data 

126 

127 

128@ router.patch( 

129 "/{name}", 

130 responses={ 

131 status.HTTP_404_NOT_FOUND: {"model": GenericError} 

132 }, 

133 status_code=status.HTTP_204_NO_CONTENT 

134) 

135def update_deployment( 

136 name: str, 

137 update: UpdateDeployment, 

138 settings: Settings = Depends(get_settings) 

139) -> Response: 

140 """Update a deployment""" 

141 deployment_path = settings.deployments_dir.joinpath(name) 

142 if not deployment_path.exists(): 

143 raise HTTPException( 

144 status_code=status.HTTP_404_NOT_FOUND, 

145 detail="Deployment does not exist" 

146 ) 

147 

148 if update.compose_file: 

149 compose_path = deployment_path.joinpath("docker-compose.yml") 

150 compose_path.write_text(update.compose_file, 

151 encoding="utf-8") 

152 

153 if update.env_file: 

154 env_path = deployment_path.joinpath(".env") 

155 env_path.write_text(update.env_file, 

156 encoding="utf-8") 

157 

158 if update.databases: 

159 db_file = deployment_path.joinpath("databases.json") 

160 db_file.touch() 

161 

162 with open(db_file, "r+", encoding="utf-8") as json_file: 

163 current_details = json.load(json_file) 

164 update_json = jsonable_encoder(update.databases) 

165 updated_details = _merge_dicts(current_details, 

166 update_json) 

167 json_file.seek(0) 

168 json.dump(updated_details, json_file, indent=4) 

169 json_file.truncate() 

170 

171 return Response(status_code=status.HTTP_204_NO_CONTENT) 

172 

173 

174@ router.delete( 

175 "/{name}", 

176 responses={ 

177 status.HTTP_404_NOT_FOUND: {"model": GenericError} 

178 }, 

179 status_code=status.HTTP_204_NO_CONTENT 

180) 

181def delete_deployment( 

182 name: str, 

183 settings: Settings = Depends(get_settings) 

184) -> Response: 

185 """Delete a deployment""" 

186 deployment_path = settings.deployments_dir.joinpath(name) 

187 if not deployment_path.exists(): 

188 raise HTTPException( 

189 status_code=status.HTTP_404_NOT_FOUND, 

190 detail="Deployment does not exist" 

191 ) 

192 rmtree(deployment_path) 

193 return Response(status_code=status.HTTP_204_NO_CONTENT) 

194 

195 

196@ router.post( 

197 "/{name}/up", 

198 responses={ 

199 status.HTTP_404_NOT_FOUND: {"model": GenericError}, 

200 status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": GenericError} 

201 }, 

202 response_model=Dict[str, str] 

203) 

204def compose_up( 

205 name: str, 

206 settings: Settings = Depends(get_settings) 

207) -> Dict[str, str]: 

208 """docker-compose up""" 

209 deployment_path = settings.deployments_dir.joinpath(name) 

210 if not deployment_path.exists(): 

211 raise HTTPException( 

212 status_code=status.HTTP_404_NOT_FOUND, 

213 detail="Deployment does not exist" 

214 ) 

215 compose_path = path.join(deployment_path, "docker-compose.yml") 

216 try: 

217 subprocess.Popen( # pylint: disable=consider-using-with 

218 f"docker-compose -f {shlex.quote(compose_path)} up -d &", 

219 shell=True, 

220 stdout=subprocess.DEVNULL, 

221 stderr=subprocess.DEVNULL 

222 ) 

223 except OSError as os_error: 

224 raise HTTPException( 

225 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

226 detail="Internal server error" 

227 ) from os_error 

228 return {"message": "docker-compose up executed"} 

229 

230 

231@ router.post( 

232 "/{name}/down", 

233 responses={ 

234 status.HTTP_404_NOT_FOUND: {"model": GenericError}, 

235 status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": GenericError} 

236 } 

237) 

238def compose_down( 

239 name: str, 

240 settings: Settings = Depends(get_settings) 

241) -> Dict[str, str]: 

242 """docker-compose down""" 

243 deployment_path = settings.deployments_dir.joinpath(name) 

244 if not deployment_path.exists(): 

245 raise HTTPException( 

246 status_code=status.HTTP_404_NOT_FOUND, 

247 detail="Deployment does not exist" 

248 ) 

249 compose_path = path.join(deployment_path, "docker-compose.yml") 

250 try: 

251 subprocess.Popen( # pylint: disable=consider-using-with 

252 f"docker-compose -f {shlex.quote(compose_path)} down &", 

253 shell=True, 

254 stdout=subprocess.DEVNULL, 

255 stderr=subprocess.DEVNULL 

256 ) 

257 except OSError as os_error: 

258 raise HTTPException( 

259 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

260 detail="Internal server error" 

261 ) from os_error 

262 return {"message": "docker-compose down executed"}