Coverage for src / idx_api / routers / admin.py: 91%

1"""Core admin endpoints for authentication and user info.""" 

2 

3from datetime import datetime 

4 

5from fastapi import APIRouter, Depends, Query 

6from pydantic import BaseModel 

7from sqlalchemy import func, select 

8from sqlalchemy.orm import Session 

9 

10from idx_api.auth import AdminUser, BrokerUser, RequiredUser 

11from idx_api.database import get_db 

12from idx_api.models.tour_request import TourRequest 

13 

14router = APIRouter() 

15 

16 

17# ===== Response Models ===== 

18 

19 

20class UserResponse(BaseModel): 

21 """Current user response.""" 

22 

23 id: int | None 

24 email: str 

25 role: str 

26 brokerage_id: int | None 

27 broker_id: int | None # Broker contact ID 

28 agent_id: int | None 

29 

30 class Config: 

31 from_attributes = True 

32 

33 

34class StatusResponse(BaseModel): 

35 """Admin status response.""" 

36 

37 authenticated: bool 

38 user: UserResponse 

39 

40 

41class TourRequestResponse(BaseModel): 

42 """Tour request response model.""" 

43 

44 id: int 

45 property_id: str 

46 first_name: str 

47 last_name: str 

48 email: str 

49 phone: str | None 

50 message: str | None 

51 preferred_date: datetime | None 

52 assigned_brokerage_id: int | None 

53 assigned_agent_id: int | None 

54 status: str 

55 created_at: datetime 

56 updated_at: datetime 

57 

58 class Config: 

59 from_attributes = True 

60 

61 

62# ===== User Info Endpoints ===== 

63 

64 

65@router.get("/me", response_model=StatusResponse) 

66async def get_current_user_info(user: RequiredUser): 

67 """ 

68 Get current authenticated user information. 

69 

70 Requires authentication (either API key or JWT token). 

71 """ 

72 return StatusResponse( 

73 authenticated=True, 

74 user=UserResponse.model_validate(user), 

75 ) 

76 

77 

78@router.get("/admin-test", response_model=dict) 

79async def admin_only_endpoint(user: AdminUser): 

80 """ 

81 Test endpoint that requires admin role. 

82 

83 Only users with admin role can access this. 

84 """ 

85 return { 

86 "message": "You have admin access!", 

87 "user_email": user.email, 

88 "user_role": user.role, 

89 } 

90 

91 

92@router.get("/broker-test", response_model=dict) 

93async def broker_endpoint(user: BrokerUser): 

94 """ 

95 Test endpoint that requires broker role (or higher). 

96 

97 Broker and admin users can access this. 

98 """ 

99 return { 

100 "message": "You have broker access!", 

101 "user_email": user.email, 

102 "user_role": user.role, 

103 "brokerage_id": user.brokerage_id, 

104 "broker_id": user.broker_id, 

105 } 

106 

107 

108# ===== Tour Listing Endpoint ===== 

109 

110 

111@router.get("/tours") 

112async def list_tours( 

113 user: AdminUser, 

114 db: Session = Depends(get_db), 

115 page: int = Query(1, ge=1), 

116 page_size: int = Query(20, ge=1, le=100), 

117): 

118 """ 

119 List all tour requests with pagination. 

120 

121 Requires admin role. 

122 """ 

123 # Count total 

124 total = db.scalar(select(func.count()).select_from(TourRequest)) 

125 

126 # Get paginated results 

127 offset = (page - 1) * page_size 

128 tours = db.scalars( 

129 select(TourRequest) 

130 .order_by(TourRequest.created_at.desc()) 

131 .offset(offset) 

132 .limit(page_size) 

133 ).all() 

134 

135 total_pages = (total + page_size - 1) // page_size 

136 

137 return { 

138 "items": [TourRequestResponse.model_validate(t) for t in tours], 

139 "total": total, 

140 "page": page, 

141 "page_size": page_size, 

142 "total_pages": total_pages, 

143 }