Coverage for /usr/lib/python3.10/site-packages/hyd/backend/version/api/v1.py: 41%

1import io 

2import os 

3import shutil 

4import tarfile 

5 

6from fastapi import ( 

7 APIRouter, 

8 Depends, 

9 Form, 

10 HTTPException, 

11 Security, 

12 UploadFile, 

13 status, 

14) 

15from sqlalchemy.orm import Session 

16 

17import hyd.backend.project.service as project_service 

18from hyd.backend.db import get_db 

19from hyd.backend.exc import ( 

20 HTTPException_UNKNOWN_PROJECT, 

21 HTTPException_UNKNOWN_VERSION, 

22 UnknownProjectError, 

23 UnknownVersionError, 

24) 

25from hyd.backend.mount_helper import MountHelper, path_to_version 

26from hyd.backend.security import Scopes 

27from hyd.backend.tag.models import TagEntry 

28from hyd.backend.user.authentication import authenticate_user 

29from hyd.backend.user.models import UserEntry 

30from hyd.backend.util.const import HEADERS 

31from hyd.backend.util.injection import inject_js_loader_to_html 

32from hyd.backend.util.logger import HydLogger 

33from hyd.backend.util.models import NameStr, PrimaryKey 

34from hyd.backend.version.models import ( 

35 API_V1_DELETE__DELETE, 

36 API_V1_LIST__GET, 

37 API_V1_UPLOAD__POST, 

38 VersionEntry, 

39 VersionResponseSchema, 

40) 

41from hyd.backend.version.service import ( 

42 create_version, 

43 delete_version_by_ref, 

44 read_version, 

45 read_versions, 

46) 

47 

48LOGGER = HydLogger("VersionAPI") 

49 

50v1_router = APIRouter(tags=["version"]) 

51 

52#################################################################################################### 

53#### HTTP Exceptions 

54#################################################################################################### 

55 

56HTTPException_EMPTY_FILE = HTTPException( 

57 status_code=status.HTTP_400_BAD_REQUEST, 

58 detail="The uploaded file has no content!", 

59 headers=HEADERS, 

60) 

61 

62#################################################################################################### 

63#### Scope: VERSION 

64#################################################################################################### 

65 

66 

67@v1_router.post("/upload", responses=API_V1_UPLOAD__POST) 

68async def _upload( 

69 file: UploadFile, 

70 project_id: PrimaryKey = Form(...), 

71 version: NameStr = Form(...), 

72 db: Session = Depends(get_db), 

73 user_entry: UserEntry = Security(authenticate_user, scopes=[Scopes.VERSION]), 

74): 

75 user_entry.check_token_project_permission(project_id=project_id) 

76 

77 try: 

78 project_entry = project_service.read_project(project_id=project_id, db=db) 

79 except UnknownProjectError: 

80 raise HTTPException_UNKNOWN_PROJECT 

81 

82 version_entry = _version_upload(file=file, project_id=project_id, version=version, db=db) 

83 

84 LOGGER.info( 

85 "{token_id: %d, user_id: %d, username: %s, project_id: %d, project_name: %s, version: %s}", 

86 user_entry.session_token_entry.id, 

87 user_entry.id, 

88 user_entry.username, 

89 project_entry.id, 

90 project_entry.name, 

91 version, 

92 ) 

93 return _version_entry_to_response_schema(version_entry) 

94 

95 

96@v1_router.get("/list", responses=API_V1_LIST__GET) 

97async def _list( 

98 project_id: PrimaryKey, 

99 db: Session = Depends(get_db), 

100 user_entry: UserEntry = Security(authenticate_user, scopes=[Scopes.VERSION]), 

101): 

102 version_entries = read_versions(project_id=project_id, db=db) 

103 return [_version_entry_to_response_schema(entry) for entry in version_entries] 

104 

105 

106@v1_router.delete("/delete", responses=API_V1_DELETE__DELETE) 

107async def _delete( 

108 project_id: PrimaryKey, 

109 version: NameStr, 

110 db: Session = Depends(get_db), 

111 user_entry: UserEntry = Security(authenticate_user, scopes=[Scopes.VERSION]), 

112): 

113 user_entry.check_token_project_permission(project_id=project_id) 

114 

115 try: 

116 version_entry = read_version(project_id=project_id, version=version, db=db) 

117 except UnknownVersionError: 

118 raise HTTPException_UNKNOWN_VERSION 

119 

120 project_entry = version_entry.project_entry 

121 version_rm_mount_and_files(version_entry=version_entry, db=db) 

122 

123 LOGGER.info( 

124 "{token_id: %d, user_id: %d, username: %s, project_id: %d, project_name: %s, version: %s}", 

125 user_entry.session_token_entry.id, 

126 user_entry.id, 

127 user_entry.username, 

128 project_entry.id, 

129 project_entry.name, 

130 version, 

131 ) 

132 

133 response = _version_entry_to_response_schema(version_entry) 

134 delete_version_by_ref(version_entry=version_entry, db=db) 

135 return response 

136 

137 

138#################################################################################################### 

139#### Util 

140#################################################################################################### 

141 

142 

143def version_rm_mount_and_files(*, version_entry: VersionEntry, db: Session) -> None: 

144 id = version_entry.project_id 

145 name = version_entry.project_entry.name 

146 version = version_entry.version 

147 

148 MountHelper.unmount_version(project_name=name, version=version) 

149 

150 tag_entries: list[TagEntry] = version_entry.tag_entries 

151 for entry in tag_entries: 

152 if entry.version: 

153 MountHelper.unmount_tag(project_name=name, tag=entry.tag) 

154 entry.version = None 

155 db.commit() 

156 

157 target = path_to_version(id, version) 

158 shutil.rmtree(target) # Delete doc files from disc 

159 

160 

161def _version_entry_to_response_schema(version_entry: VersionEntry) -> VersionResponseSchema: 

162 tag_entries: list[TagEntry] = version_entry.tag_entries 

163 

164 return VersionResponseSchema( 

165 project_id=version_entry.project_id, 

166 version=version_entry.version, 

167 created_at=version_entry.created_at, 

168 tags=[t_entry.tag for t_entry in tag_entries], 

169 ) 

170 

171 

172def _version_upload( 

173 *, file: UploadFile, project_id: PrimaryKey, version: NameStr, db: Session 

174) -> VersionEntry: 

175 

176 file_content = file.file.read() 

177 if not file_content: 

178 raise HTTPException_EMPTY_FILE 

179 

180 version_entry = create_version( 

181 project_id=project_id, 

182 version=version, 

183 filename=file.filename, 

184 content_type=file.content_type, 

185 db=db, 

186 ) 

187 

188 # Extract doc files to disc 

189 file_like_object = io.BytesIO(file_content) 

190 tar = tarfile.open(fileobj=file_like_object, mode="r:gz") 

191 target = path_to_version(version_entry.project_id, version_entry.version) 

192 os.makedirs(target, exist_ok=True) 

193 tar.extractall(target) 

194 

195 inject_js_loader_to_html(dir_path=target) 

196 

197 MountHelper.mount_version(version_entry=version_entry) 

198 

199 return version_entry