Coverage for routes\events.py: 90%

41 statements  

« prev     ^ index     » next       coverage.py v7.1.0, created at 2023-02-05 19:00 +0800

1from typing import List 

2 

3from auth.authenticate import authenticate 

4from beanie import PydanticObjectId 

5from database.connection import Database 

6from fastapi import APIRouter, Depends, HTTPException, status 

7from models.events import Event, EventUpdate 

8 

9event_router = APIRouter( 

10 tags=["Events"] 

11) 

12 

13event_database = Database(Event) 

14 

15 

16@event_router.get("/", response_model=List[Event]) 

17async def retrieve_all_events() -> List[Event]: 

18 events = await event_database.get_all() 

19 return events 

20 

21 

22@event_router.get("/{id}", response_model=Event) 

23async def retrieve_event(id: PydanticObjectId) -> Event: 

24 event = await event_database.get(id) 

25 if not event: 

26 raise HTTPException( 

27 status_code=status.HTTP_404_NOT_FOUND, 

28 detail="Event with supplied ID does not exist" 

29 ) 

30 return event 

31 

32 

33@event_router.post("/new") 

34async def create_event(body: Event, user: str = Depends(authenticate)) -> dict: 

35 body.creator = user 

36 await event_database.save(body) 

37 return { 

38 "message": "Event created successfully" 

39 } 

40 

41 

42@event_router.put("/{id}", response_model=Event) 

43async def update_event(id: PydanticObjectId, body: EventUpdate, user: str = Depends(authenticate)) -> Event: 

44 event = await event_database.get(id) 

45 if event.creator != user: 

46 raise HTTPException( 

47 status_code=status.HTTP_400_BAD_REQUEST, 

48 detail="Operation not allowed" 

49 ) 

50 updated_event = await event_database.update(id, body) 

51 if not updated_event: 

52 raise HTTPException( 

53 status_code=status.HTTP_404_NOT_FOUND, 

54 detail="Event with supplied ID does not exist" 

55 ) 

56 return updated_event 

57 

58 

59@event_router.delete("/{id}") 

60async def delete_event(id: PydanticObjectId, user: str = Depends(authenticate)) -> dict: 

61 event = await event_database.get(id) 

62 if event.creator != user: 

63 raise HTTPException( 

64 status_code=status.HTTP_404_NOT_FOUND, 

65 detail="Event not found" 

66 ) 

67 if not event: 

68 raise HTTPException( 

69 status_code=status.HTTP_404_NOT_FOUND, 

70 detail="Event with supplied ID does not exist" 

71 ) 

72 await event_database.delete(id) 

73 

74 return { 

75 "message": "Event deleted successfully." 

76 }