Coverage for app/logic/celtsLabor.py: 81%

1import requests 

2import json 

3from peewee import DoesNotExist 

4from collections import defaultdict 

5from app import app 

6from app.models.user import User 

7from app.models.celtsLabor import CeltsLabor 

8from app.models.term import Term 

9 

10def getCeltsLaborFromLsf(): 

11 """ 

12 Make a call to the LSF endpoint which returns all CELTS student labor records.  

13 

14 The returned data is a dictionary with B# key and value that is a list of dicts that contain the labor information.  

15 

16 """ 

17 try: 

18 lsfUrl = f"{app.config['lsf_url'].strip('/')}/api/org/2084" 

19 response = requests.get(lsfUrl) 

20 return response.json() 

21 except json.decoder.JSONDecodeError: 

22 print(f'Response from {lsfUrl} was not JSON.\n' + response.text) 

23 return {} 

24 except KeyError as e: 

25 print(f'Make sure you have "lsf_url" set in your local-override config file.') 

26 raise(e) 

27 

28def updateCeltsLaborFromLsf(): 

29 """ 

30 Parse the LSF response object so that it is only labor records for summer and academic years. 

31 

32 Input: JSON response 

33 {'B#': [{'positionTitle': 'Position Title', 

34 'termCode': 'Term Code',  

35 'laborStart': Labor Start Date,  

36 'laborEnd': Labor End Date,  

37 'jobType': 'Job Type', 

38 'wls': 'WLS Lvl', 

39 'termName': 'Term Name'}], 

40 'B#': [{'positionTitle': 'Position Title', 

41 'termCode': 'Term Code',  

42 'laborStart': Labor Start Date,  

43 'laborEnd': Labor End Date,  

44 'jobType': 'Job Type', 

45 'wls': 'WLS Lvl', 

46 'termName': 'Term Name'}]} 

47 """ 

48 laborDict = getCeltsLaborFromLsf() 

49 

50 studentLaborDict = {} 

51 for key, value in laborDict.items(): 

52 try: 

53 username = User.get(bnumber = key) 

54 # All term codes for summer end with 13 and all term codes for an academic year end in 00 and those are the only terms we want to record. 

55 studentLaborDict[username] = collapsePositions([p for p in value if str(p["termCode"])[-2:] in ["00","13"]]) 

56 except DoesNotExist: 

57 pass 

58 

59 refreshCeltsLaborRecords(studentLaborDict) 

60 

61def collapsePositions(positionList): 

62 ''' 

63 Parse position information from the JSON response and return only the position title and term title  

64 

65 Input: [{'positionTitle': 'Position Title', 

66 'termCode': 'Term Code',  

67 'laborStart': Labor Start Date,  

68 'laborEnd': Labor End Date,  

69 'jobType': 'Job Type', 

70 'wls': 'WLS Lvl', 

71 'termName': 'Term Name'}] 

72  

73 Returned:{'Position Title': ['Term Name']} 

74 

75 ''' 

76 laborPositionDict = defaultdict(list) 

77 

78 for position in positionList: 

79 laborPositionDict[position['positionTitle']].append(position['termName'].replace("AY ", "")) 

80 

81 return laborPositionDict 

82 

83def refreshCeltsLaborRecords(laborDict): 

84 """ 

85 Input: Dictionary containing CELTS labor information 

86 {'username1': {'positionTitle': ['termName'], 

87 'positionTitle': ['termName']}, 

88 'username2': {'positionTitle': ['termName', 'termName']} 

89 } 

90 

91 Delete records for the students that are currently in the CeltsLabor table 

92 if they are also in the laborDict and save content of laborDict to the table.  

93 """ 

94 

95 celtsLabor = [] 

96 for key, value in laborDict.items(): 

97 for positionTitle, termNames in value.items(): 

98 for term in termNames: 

99 termTableMatch = Term.select() 

100 if term[0].isalpha(): 

101 termTableMatch = termTableMatch.where(Term.description == term) 

102 else: 

103 termTableMatch = termTableMatch.where(Term.academicYear == term, Term.description % "Fall%") 

104 try: 

105 laborTerm = termTableMatch.get() 

106 isAcademicYear = not laborTerm.isSummer 

107 celtsLabor.append({"user": key, 

108 "positionTitle": positionTitle, 

109 "term": laborTerm, 

110 "isAcademicYear": isAcademicYear}) 

111 except DoesNotExist: 

112 pass 

113 

114 CeltsLabor.delete().where(CeltsLabor.user << [username['user'] for username in celtsLabor]).execute() 

115 CeltsLabor.insert_many(celtsLabor).on_conflict_replace().execute() 

116 

117def getCeltsLaborHistory(volunteer): 

118 

119 laborHistoryList = list(CeltsLabor.select(CeltsLabor.positionTitle, 

120 CeltsLabor.id, 

121 Term.description, 

122 Term.academicYear, 

123 Term.isSummer) 

124 .join(Term, on=(CeltsLabor.term == Term.id)) 

125 .where(CeltsLabor.user == volunteer) 

126 .order_by(Term.termOrder.asc())) 

127 termsByAcademicYear = {} 

128 for position in laborHistoryList: 

129 if position.term.isSummer: 

130 continue 

131 academicYear = position.term.academicYear 

132 description = position.term.description 

133 if academicYear not in termsByAcademicYear: 

134 termsByAcademicYear[academicYear] = {"Fall": False,"Spring": False} 

135 if "Fall" in description: 

136 termsByAcademicYear[academicYear]["Fall"] = True 

137 elif "Spring" in description: 

138 termsByAcademicYear[academicYear]["Spring"] = True 

139 laborHistoryDict= {} 

140 for position in laborHistoryList: 

141 description = position.term.description 

142 academicYear = position.term.academicYear 

143 if position.term.isSummer: 

144 positionTerm = description 

145 else: 

146 hasFall = termsByAcademicYear[academicYear]["Fall"] 

147 hasSpring = termsByAcademicYear[academicYear]["Spring"] 

148 if hasFall and hasSpring: 

149 positionTerm = description 

150 else: 

151 positionTerm = f"AY {academicYear}" 

152 laborHistoryDict[position.id] = (position.positionTitle,positionTerm) 

153 return laborHistoryDict