Coverage for backend/app/routers/data_tables.py: 87%

1"""Module for generating CRUD routers for the JAM data tables""" 

2 

3import base64 

4 

5from fastapi import Depends, status, HTTPException, Response 

6from sqlalchemy.orm import Session 

7 

8from app import models, database, oauth2, schemas 

9from app.routers import generate_data_table_crud_router 

10 

11# Settings router 

12settings_router = generate_data_table_crud_router( 

13 table_model=models.Setting, 

14 create_schema=schemas.SettingCreate, 

15 update_schema=schemas.SettingUpdate, 

16 out_schema=schemas.SettingOut, 

17 endpoint="settings", 

18 not_found_msg="Setting not found", 

19 admin_only=True, 

20) 

21 

22# Keyword router 

23keyword_router = generate_data_table_crud_router( 

24 table_model=models.Keyword, 

25 create_schema=schemas.KeywordCreate, 

26 update_schema=schemas.KeywordUpdate, 

27 out_schema=schemas.KeywordOut, 

28 endpoint="keywords", 

29 not_found_msg="Keyword not found", 

30) 

31 

32# Aggregator router 

33aggregator_router = generate_data_table_crud_router( 

34 table_model=models.Aggregator, 

35 create_schema=schemas.AggregatorCreate, 

36 update_schema=schemas.AggregatorUpdate, 

37 out_schema=schemas.AggregatorOut, 

38 endpoint="aggregators", 

39 not_found_msg="Aggregator not found", 

40) 

41 

42# Company router 

43company_router = generate_data_table_crud_router( 

44 table_model=models.Company, 

45 create_schema=schemas.CompanyCreate, 

46 update_schema=schemas.CompanyUpdate, 

47 out_schema=schemas.CompanyOut, 

48 endpoint="companies", 

49 not_found_msg="Company not found", 

50) 

51 

52# Location router 

53location_router = generate_data_table_crud_router( 

54 table_model=models.Location, 

55 create_schema=schemas.LocationCreate, 

56 update_schema=schemas.LocationUpdate, 

57 out_schema=schemas.LocationOut, 

58 endpoint="locations", 

59 not_found_msg="Location not found", 

60) 

61 

62# Person router 

63person_router = generate_data_table_crud_router( 

64 table_model=models.Person, 

65 create_schema=schemas.PersonCreate, 

66 update_schema=schemas.PersonUpdate, 

67 out_schema=schemas.PersonOut, 

68 endpoint="persons", 

69 not_found_msg="Person not found", 

70) 

71 

72# Job router 

73job_router = generate_data_table_crud_router( 

74 table_model=models.Job, 

75 create_schema=schemas.JobCreate, 

76 update_schema=schemas.JobUpdate, 

77 out_schema=schemas.JobOut, 

78 endpoint="jobs", 

79 not_found_msg="Job not found", 

80 many_to_many_fields={ 

81 "keywords": {"table": models.job_keyword_mapping, "local_key": "job_id", "remote_key": "keyword_id"}, 

82 "contacts": {"table": models.job_contact_mapping, "local_key": "job_id", "remote_key": "person_id"}, 

83 }, 

84) 

85 

86# Interview router 

87interview_router = generate_data_table_crud_router( 

88 table_model=models.Interview, 

89 create_schema=schemas.InterviewCreate, 

90 update_schema=schemas.InterviewUpdate, 

91 out_schema=schemas.InterviewOut, 

92 endpoint="interviews", 

93 not_found_msg="Interview not found", 

94 many_to_many_fields={ 

95 "interviewers": { 

96 "table": models.interview_interviewer_mapping, 

97 "local_key": "interview_id", 

98 "remote_key": "person_id", 

99 }, 

100 }, 

101) 

102 

103# Job Application Update router 

104job_application_update_router = generate_data_table_crud_router( 

105 table_model=models.JobApplicationUpdate, 

106 create_schema=schemas.JobApplicationUpdateCreate, 

107 update_schema=schemas.JobApplicationUpdateUpdate, 

108 out_schema=schemas.JobApplicationUpdateOut, 

109 endpoint="jobapplicationupdates", 

110 not_found_msg="Job Application Update not found", 

111) 

112 

113# File router 

114file_router = generate_data_table_crud_router( 

115 table_model=models.File, 

116 create_schema=schemas.FileCreate, 

117 update_schema=schemas.FileUpdate, 

118 out_schema=schemas.FileOut, 

119 endpoint="files", 

120 not_found_msg="File not found", 

121) 

122 

123 

124@file_router.get("/{file_id}/download") 

125def download_file( 

126 file_id: int, 

127 db: Session = Depends(database.get_db), 

128 current_user: models.User = Depends(oauth2.get_current_user), 

129): 

130 """Download a file by ID. 

131 :param file_id: The file ID. 

132 :param db: The database session. 

133 :param current_user: The current user.""" 

134 

135 # Get file record from the database 

136 # noinspection PyTypeChecker 

137 file_record = ( 

138 db.query(models.File).filter(models.File.id == file_id, models.File.owner_id == current_user.id).first() 

139 ) 

140 

141 if not file_record: 

142 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") 

143 

144 if not file_record.content: 

145 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File content not found") 

146 

147 try: 

148 # Content is already a base64 string - just decode it 

149 if file_record.content.startswith("data:"): 

150 encoded_data = file_record.content.split(",", 1)[1] 

151 file_content = base64.b64decode(encoded_data) 

152 else: 

153 # Pure base64 string 

154 file_content = base64.b64decode(file_record.content) 

155 

156 except Exception as e: 

157 raise HTTPException( 

158 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error decoding file content: {str(e)}" 

159 ) 

160 

161 content_type = file_record.type if file_record.type else "application/octet-stream" 

162 

163 return Response( 

164 content=file_content, 

165 media_type=content_type, 

166 headers={ 

167 "Content-Disposition": f'attachment; filename="{file_record.filename}"', 

168 "Content-Length": str(len(file_content)), 

169 }, 

170 )