Coverage for src / idx_api / routers / suggestions.py: 39%
141 statements
« prev ^ index » next coverage.py v7.13.1, created at 2025-12-28 11:09 -0700
1"""Suggestions API for suggest-edit workflow."""
3from datetime import datetime
4from typing import Any
6from fastapi import APIRouter, Depends, HTTPException, Query
7from pydantic import BaseModel
8from sqlalchemy import func, select
9from sqlalchemy.orm import Session
11from idx_api.auth import AdminUser, BrokerUser, RequiredUser
12from idx_api.database import get_db
13from idx_api.models.suggestion import Suggestion
15router = APIRouter()
18# ===== Request/Response Models =====
21class SuggestionCreate(BaseModel):
22 """Create suggestion request."""
24 target_table: str
25 target_id: int
26 suggestion_type: str
27 title: str
28 description: str | None = None
29 proposed_changes: dict[str, Any]
30 reviewer_type: str # "admin" or "broker"
33class SuggestionReject(BaseModel):
34 """Reject suggestion request."""
36 notes: str
39class SuggestionResponse(BaseModel):
40 """Suggestion response model."""
42 id: int
43 target_table: str
44 target_id: int
45 suggestion_type: str
46 title: str
47 description: str | None
48 proposed_changes: dict[str, Any]
49 reviewer_type: str
50 status: str
51 submitted_by_id: int | None
52 reviewed_by_id: int | None
53 review_notes: str | None
54 created_at: datetime
55 reviewed_at: datetime | None
57 class Config:
58 from_attributes = True
61class PaginatedSuggestions(BaseModel):
62 """Paginated suggestions response."""
64 items: list[SuggestionResponse]
65 total: int
66 page: int
67 page_size: int
68 total_pages: int
71# ===== Endpoints =====
74@router.get("/suggestions", response_model=PaginatedSuggestions)
75async def list_suggestions(
76 user: RequiredUser,
77 db: Session = Depends(get_db),
78 page: int = Query(1, ge=1),
79 page_size: int = Query(20, ge=1, le=100),
80 status: str | None = None,
81 target_table: str | None = None,
82):
83 """
84 List suggestions with filtering.
86 Filters results based on user role:
87 - Admins: See all suggestions
88 - Brokers: See suggestions for their brokerage
89 - Agents: See suggestions they submitted
90 """
91 # Build query
92 query = select(Suggestion)
94 # Apply filters
95 if status:
96 query = query.where(Suggestion.status == status)
97 if target_table:
98 query = query.where(Suggestion.target_table == target_table)
100 # Role-based filtering
101 if user.role == "broker" and user.broker_id:
102 # Brokers only see suggestions for their brokerage resources
103 query = query.where(
104 (Suggestion.target_table == "broker") & (Suggestion.target_id == user.broker_id)
105 | (Suggestion.target_table == "agent") # All agent suggestions for their brokerage
106 )
107 elif user.role == "agent":
108 # Agents only see their own suggestions
109 query = query.where(Suggestion.submitted_by_id == user.id)
111 # Count total
112 count_query = select(func.count()).select_from(Suggestion)
113 if status:
114 count_query = count_query.where(Suggestion.status == status)
115 if target_table:
116 count_query = count_query.where(Suggestion.target_table == target_table)
117 total = db.scalar(count_query)
119 # Get paginated results
120 offset = (page - 1) * page_size
121 suggestions = db.scalars(
122 query.order_by(Suggestion.created_at.desc()).offset(offset).limit(page_size)
123 ).all()
125 total_pages = (total + page_size - 1) // page_size
127 return PaginatedSuggestions(
128 items=[SuggestionResponse.model_validate(s) for s in suggestions],
129 total=total,
130 page=page,
131 page_size=page_size,
132 total_pages=total_pages,
133 )
136@router.post("/suggestions", response_model=SuggestionResponse)
137async def create_suggestion(
138 data: SuggestionCreate,
139 user: RequiredUser,
140 db: Session = Depends(get_db),
141):
142 """
143 Submit a new suggestion.
145 Any authenticated user can submit suggestions.
146 """
147 suggestion = Suggestion(
148 target_table=data.target_table,
149 target_id=data.target_id,
150 suggestion_type=data.suggestion_type,
151 title=data.title,
152 description=data.description,
153 proposed_changes=data.proposed_changes,
154 reviewer_type=data.reviewer_type,
155 status="pending",
156 submitted_by_id=user.id,
157 created_at=datetime.utcnow(),
158 )
160 db.add(suggestion)
161 db.commit()
162 db.refresh(suggestion)
164 return SuggestionResponse.model_validate(suggestion)
167@router.get("/suggestions/{suggestion_id}", response_model=SuggestionResponse)
168async def get_suggestion(
169 suggestion_id: int,
170 user: RequiredUser,
171 db: Session = Depends(get_db),
172):
173 """Get a specific suggestion."""
174 suggestion = db.get(Suggestion, suggestion_id)
175 if not suggestion:
176 raise HTTPException(status_code=404, detail="Suggestion not found")
178 # Check permissions
179 if user.role == "broker" and user.broker_id:
180 # Brokers can only see suggestions for their resources
181 if suggestion.target_table == "broker" and suggestion.target_id != user.broker_id:
182 raise HTTPException(status_code=403, detail="Not authorized to view this suggestion")
183 elif user.role == "agent":
184 # Agents can only see their own suggestions
185 if suggestion.submitted_by_id != user.id:
186 raise HTTPException(status_code=403, detail="Not authorized to view this suggestion")
188 return SuggestionResponse.model_validate(suggestion)
191@router.post("/suggestions/{suggestion_id}/approve", response_model=SuggestionResponse)
192async def approve_suggestion(
193 suggestion_id: int,
194 user: RequiredUser,
195 db: Session = Depends(get_db),
196):
197 """
198 Approve and apply a suggestion.
200 Requires admin role or broker role (for agent suggestions in their brokerage).
201 """
202 suggestion = db.get(Suggestion, suggestion_id)
203 if not suggestion:
204 raise HTTPException(status_code=404, detail="Suggestion not found")
206 if suggestion.status != "pending":
207 raise HTTPException(status_code=400, detail="Suggestion already reviewed")
209 # Check permissions
210 if suggestion.reviewer_type == "admin":
211 # Only admins can approve admin-level suggestions
212 if user.role != "admin":
213 raise HTTPException(status_code=403, detail="Admin approval required")
214 elif suggestion.reviewer_type == "broker":
215 # Brokers can approve suggestions for their brokerage
216 if user.role == "broker" and user.broker_id:
217 # Verify the suggestion is for their brokerage
218 if suggestion.target_table == "agent":
219 # Need to check if the agent belongs to their brokerage
220 from idx_api.models.agent import Agent
221 agent = db.get(Agent, suggestion.target_id)
222 if not agent or agent.broker_id != user.broker_id:
223 raise HTTPException(status_code=403, detail="Not authorized")
224 elif user.role != "admin":
225 raise HTTPException(status_code=403, detail="Broker or admin approval required")
227 # Apply the proposed changes
228 # NOTE: This is a simplified implementation - production would need to:
229 # 1. Validate the proposed changes
230 # 2. Apply them to the target table
231 # 3. Handle field-specific validation
232 # For now, we'll just mark as approved - actual application would be in Phase 4
234 suggestion.status = "approved"
235 suggestion.reviewed_by_id = user.id
236 suggestion.reviewed_at = datetime.utcnow()
238 db.commit()
239 db.refresh(suggestion)
241 return SuggestionResponse.model_validate(suggestion)
244@router.post("/suggestions/{suggestion_id}/reject", response_model=SuggestionResponse)
245async def reject_suggestion(
246 suggestion_id: int,
247 data: SuggestionReject,
248 user: RequiredUser,
249 db: Session = Depends(get_db),
250):
251 """
252 Reject a suggestion.
254 Requires admin role or broker role (for agent suggestions).
255 """
256 suggestion = db.get(Suggestion, suggestion_id)
257 if not suggestion:
258 raise HTTPException(status_code=404, detail="Suggestion not found")
260 if suggestion.status != "pending":
261 raise HTTPException(status_code=400, detail="Suggestion already reviewed")
263 # Check permissions (same logic as approve)
264 if suggestion.reviewer_type == "admin" and user.role != "admin":
265 raise HTTPException(status_code=403, detail="Admin approval required")
266 elif suggestion.reviewer_type == "broker":
267 if user.role == "broker" and user.broker_id:
268 from idx_api.models.agent import Agent
269 if suggestion.target_table == "agent":
270 agent = db.get(Agent, suggestion.target_id)
271 if not agent or agent.broker_id != user.broker_id:
272 raise HTTPException(status_code=403, detail="Not authorized")
273 elif user.role != "admin":
274 raise HTTPException(status_code=403, detail="Broker or admin approval required")
276 suggestion.status = "rejected"
277 suggestion.reviewed_by_id = user.id
278 suggestion.review_notes = data.notes
279 suggestion.reviewed_at = datetime.utcnow()
281 db.commit()
282 db.refresh(suggestion)
284 return SuggestionResponse.model_validate(suggestion)
287@router.delete("/suggestions/{suggestion_id}")
288async def delete_suggestion(
289 suggestion_id: int,
290 user: AdminUser,
291 db: Session = Depends(get_db),
292):
293 """
294 Delete a suggestion.
296 Admin only.
297 """
298 suggestion = db.get(Suggestion, suggestion_id)
299 if not suggestion:
300 raise HTTPException(status_code=404, detail="Suggestion not found")
302 db.delete(suggestion)
303 db.commit()
305 return {"message": "Suggestion deleted successfully"}