Coverage for model/announcement.py: 96%

1from datetime import datetime 

2from typing import Optional 

3from fastapi import APIRouter, Depends 

4 

5from mongo import * 

6from mongo.utils import * 

7from .auth import login_required 

8from .utils import * 

9from .schemas import CreateAnnouncementBody, UpdateAnnouncementBody, DeleteAnnouncementBody 

10from .course import course_router 

11 

12__all__ = ['ann_router'] 

13 

14ann_router = APIRouter() 

15 

16 

17def _format_ann(an): 

18 return { 

19 'annId': str(an.id), 

20 'title': an.title, 

21 'createTime': int(an.create_time.timestamp()), 

22 'updateTime': int(an.update_time.timestamp()), 

23 'creator': an.creator.info, 

24 'updater': an.updater.info, 

25 'markdown': an.markdown, 

26 'pinned': an.pinned, 

27 } 

28 

29 

30@ann_router.get('') 

31@ann_router.get('/{ann_id}') 

32def get_sys_ann(ann_id: Optional[str] = None): 

33 public_name = Course.get_public().course_name 

34 anns = Announcement.ann_list(None, public_name) 

35 data = [ 

36 _format_ann(an) for an in anns 

37 if ann_id is None or str(an.id) == ann_id 

38 ] 

39 return HTTPResponse('Sys Ann bro', data=data) 

40 

41 

42@course_router.get('/{course_name}/ann') 

43def get_course_announcements(course_name: str, user=Depends(login_required)): 

44 try: 

45 anns = Announcement.ann_list(user.obj, course_name) 

46 except (DoesNotExist, ValidationError): 

47 return HTTPError('Cannot Access a Announcement', 403) 

48 if anns is None: 

49 return HTTPError('Announcement Not Found', 404) 

50 data = [_format_ann(an) for an in anns] 

51 return HTTPResponse('Announcement List', data=data) 

52 

53 

54@ann_router.get('/{course_name}/{ann_id}') 

55def get_ann_by_id(course_name: str, ann_id: str, user=Depends(login_required)): 

56 try: 

57 anns = Announcement.ann_list(user.obj, course_name) 

58 except (DoesNotExist, ValidationError): 

59 return HTTPError('Cannot Access a Announcement', 403) 

60 if anns is None: 

61 return HTTPError('Announcement Not Found', 404) 

62 data = [_format_ann(an) for an in anns if str(an.id) == ann_id] 

63 return HTTPResponse('Announcement List', data=data) 

64 

65 

66@ann_router.post('') 

67def create_announcement(body: CreateAnnouncementBody, 

68 user=Depends(login_required)): 

69 try: 

70 ann = Announcement.new_ann( 

71 title=body.title, 

72 creator=user.obj, 

73 markdown=body.markdown, 

74 pinned=body.pinned, 

75 course=body.course_name or 'Public', 

76 ) 

77 except ValidationError as ve: 

78 return HTTPError('Failed to Create Announcement', 

79 400, 

80 data=ve.to_dict()) 

81 if ann is None: 

82 return HTTPError('Failed to Create Announcement', 403) 

83 data = { 

84 'annId': str(ann.id), 

85 'createTime': int(ann.create_time.timestamp()), 

86 } 

87 return HTTPResponse('Announcement Created', data=data) 

88 

89 

90@ann_router.put('') 

91def update_announcement(body: UpdateAnnouncementBody, 

92 user=Depends(login_required)): 

93 ann = Announcement(body.ann_id) 

94 if not ann: 

95 return HTTPError('Announcement Not Found', 404) 

96 course = Course(ann.course) 

97 if not course.permission(user, Course.Permission.GRADE): 

98 return HTTPError('Failed to Update Announcement', 403) 

99 try: 

100 ann.update( 

101 title=body.title, 

102 markdown=body.markdown, 

103 update_time=datetime.now(), 

104 updater=user.obj, 

105 pinned=body.pinned, 

106 ) 

107 except ValidationError as ve: 

108 return HTTPError('Failed to Update Announcement', 

109 400, 

110 data=ve.to_dict()) 

111 return HTTPResponse('Updated') 

112 

113 

114@ann_router.delete('') 

115def delete_announcement(body: DeleteAnnouncementBody, 

116 user=Depends(login_required)): 

117 ann = Announcement(body.ann_id) 

118 if not ann: 

119 return HTTPError('Announcement Not Found', 404) 

120 course = Course(ann.course) 

121 if not course.permission(user, Course.Permission.GRADE): 

122 return HTTPError('Failed to Delete Announcement', 403) 

123 ann.update(status=1) 

124 return HTTPResponse('Deleted')