Coverage for src / mysingle / auth / router / users.py: 0%

1from beanie import PydanticObjectId 

2from fastapi import APIRouter, Request, Response, status 

3 

4from ..deps import admin_only, get_current_active_verified_user 

5from ..exceptions import ( 

6 UserNotExists, 

7) 

8from ..schemas import UserResponse, UserUpdate 

9from ..user_manager import UserManager 

10 

11user_manager = UserManager() 

12 

13 

14def get_users_router() -> APIRouter: 

15 """Generate a router with the authentication routes.""" 

16 router = APIRouter() 

17 

18 @router.get( 

19 "/me", 

20 response_model=UserResponse, 

21 ) 

22 async def get_user_me( 

23 request: Request, 

24 ) -> UserResponse: 

25 # Request에서 인증된 사용자 가져오기 

26 current_user = get_current_active_verified_user(request) 

27 return UserResponse(**current_user.model_dump(by_alias=True)) 

28 

29 @router.get( 

30 "/me/activity", 

31 response_model=dict, 

32 ) 

33 async def get_user_activity( 

34 request: Request, 

35 ) -> dict: 

36 """ 

37 현재 사용자의 활동 기록 조회. 

38 

39 Returns: 

40 dict: 사용자 활동 요약 정보 

41 """ 

42 current_user = get_current_active_verified_user(request) 

43 return await user_manager.get_user_activity_summary(current_user) 

44 

45 @router.patch( 

46 "/me", 

47 response_model=UserResponse, 

48 ) 

49 async def update_user_me( 

50 request: Request, 

51 obj_in: UserUpdate, 

52 ) -> UserResponse: 

53 current_user = get_current_active_verified_user(request) 

54 user = await user_manager.update(obj_in, current_user) 

55 return UserResponse(**user.model_dump(by_alias=True)) 

56 

57 @router.get( 

58 "/{user_id}", 

59 response_model=UserResponse, 

60 ) 

61 @admin_only 

62 async def get_user( 

63 user_id: PydanticObjectId, 

64 ) -> UserResponse: 

65 user = await user_manager.get(user_id) 

66 if user is None: 

67 raise UserNotExists(identifier=str(user_id), identifier_type="user") 

68 return UserResponse(**user.model_dump(by_alias=True)) 

69 

70 @router.get( 

71 "/{user_id}/activity", 

72 response_model=dict, 

73 ) 

74 @admin_only 

75 async def get_user_activity_by_id( 

76 user_id: PydanticObjectId, 

77 ) -> dict: 

78 """ 

79 특정 사용자의 활동 기록 조회 (관리자 전용). 

80 

81 Args: 

82 id: 조회할 사용자 ID 

83 

84 Returns: 

85 dict: 사용자 활동 요약 정보 

86 """ 

87 user = await user_manager.get(user_id) 

88 if user is None: 

89 raise UserNotExists(identifier=str(user_id), identifier_type="user") 

90 return await user_manager.get_user_activity_summary(user) 

91 

92 @router.patch( 

93 "/{user_id}", 

94 response_model=UserResponse, 

95 ) 

96 @admin_only 

97 async def update_user( 

98 request: Request, 

99 user_id: PydanticObjectId, 

100 obj_in: UserUpdate, # type: ignore 

101 ) -> UserResponse: 

102 user = await user_manager.get(user_id) 

103 if user is None: 

104 raise UserNotExists(identifier=str(user_id), identifier_type="user") 

105 updated_user = await user_manager.update(obj_in, user, request=request) 

106 return UserResponse(**updated_user.model_dump(by_alias=True)) 

107 

108 @router.delete( 

109 "/{user_id}", 

110 status_code=status.HTTP_204_NO_CONTENT, 

111 response_class=Response, 

112 ) 

113 @admin_only 

114 async def delete_user( 

115 request: Request, 

116 user_id: PydanticObjectId, 

117 ) -> None: 

118 user = await user_manager.get(user_id) 

119 if user is None: 

120 raise UserNotExists(identifier=str(user_id), identifier_type="user") 

121 await user_manager.delete(user, request=request) 

122 return None 

123 

124 return router