Coverage for src / idx_api / routers / properties.py: 17%

1"""Property listing endpoints.""" 

2 

3import json 

4import os 

5import time 

6from pathlib import Path 

7 

8from fastapi import APIRouter, Depends, HTTPException, Query 

9from fastapi.responses import Response 

10from sqlalchemy import text 

11from sqlalchemy.orm import Session 

12 

13from idx_api.database import get_db 

14from idx_api.pdf_generator import generate_property_pdf 

15from idx_api.schemas.property import ( 

16 PropertyDetail, 

17 PropertyListItem, 

18 PropertySearchParams, 

19 PropertySearchResponse, 

20) 

21 

22router = APIRouter() 

23 

24 

25def load_site_config(): 

26 """Load site configuration from site.json.""" 

27 # Look for site.json in the web project 

28 site_json_path = Path(__file__).parent.parent.parent.parent / "idx-web" / "src" / "config" / "site.json" 

29 

30 if site_json_path.exists(): 

31 with open(site_json_path, "r") as f: 

32 return json.load(f) 

33 

34 # Fallback to empty config if not found 

35 return { 

36 "site": {"name": "IDX Property Listings"}, 

37 "contact": {}, 

38 "branding": {} 

39 } 

40 

41 

42@router.get("", response_model=PropertySearchResponse) 

43async def list_properties( 

44 params: PropertySearchParams = Depends(), 

45 db: Session = Depends(get_db), 

46): 

47 """ 

48 List properties with filtering and pagination. 

49 

50 Supports: 

51 - Location filters (city, state, postal code) 

52 - Bounding box for map view 

53 - Radius search from a point 

54 - Property attribute filters 

55 """ 

56 start_time = time.perf_counter() 

57 

58 # Build WHERE clauses 

59 conditions = ["mlg_can_view = 1"] 

60 bind_params = {} 

61 

62 if params.status: 

63 conditions.append("standard_status = :status") 

64 bind_params["status"] = params.status 

65 

66 if params.city: 

67 conditions.append("city = :city") 

68 bind_params["city"] = params.city 

69 

70 if params.state_or_province: 

71 conditions.append("state_or_province = :state") 

72 bind_params["state"] = params.state_or_province 

73 

74 if params.postal_code: 

75 conditions.append("postal_code = :postal") 

76 bind_params["postal"] = params.postal_code 

77 

78 if params.property_type: 

79 conditions.append("property_type = :prop_type") 

80 bind_params["prop_type"] = params.property_type 

81 

82 if params.min_price: 

83 conditions.append("list_price >= :min_price") 

84 bind_params["min_price"] = params.min_price 

85 

86 if params.max_price: 

87 conditions.append("list_price <= :max_price") 

88 bind_params["max_price"] = params.max_price 

89 

90 if params.min_beds: 

91 conditions.append("bedrooms_total >= :min_beds") 

92 bind_params["min_beds"] = params.min_beds 

93 

94 if params.max_beds: 

95 conditions.append("bedrooms_total <= :max_beds") 

96 bind_params["max_beds"] = params.max_beds 

97 

98 if params.min_baths: 

99 conditions.append("bathrooms_total_integer >= :min_baths") 

100 bind_params["min_baths"] = params.min_baths 

101 

102 if params.min_sqft: 

103 conditions.append("living_area >= :min_sqft") 

104 bind_params["min_sqft"] = params.min_sqft 

105 

106 if params.max_sqft: 

107 conditions.append("living_area <= :max_sqft") 

108 bind_params["max_sqft"] = params.max_sqft 

109 

110 if params.min_year: 

111 conditions.append("year_built >= :min_year") 

112 bind_params["min_year"] = params.min_year 

113 

114 if params.max_year: 

115 conditions.append("year_built <= :max_year") 

116 bind_params["max_year"] = params.max_year 

117 

118 # Bounding box search using R*Tree 

119 geo_join = "" 

120 if all([params.min_lat, params.max_lat, params.min_lng, params.max_lng]): 

121 geo_join = "JOIN property_geo g ON p.id = g.property_id" 

122 conditions.append("g.min_lng >= :min_lng AND g.max_lng <= :max_lng") 

123 conditions.append("g.min_lat >= :min_lat AND g.max_lat <= :max_lat") 

124 bind_params.update({ 

125 "min_lat": params.min_lat, 

126 "max_lat": params.max_lat, 

127 "min_lng": params.min_lng, 

128 "max_lng": params.max_lng, 

129 }) 

130 

131 # Radius search (uses bounding box + Haversine) 

132 distance_select = "" 

133 if params.center_lat and params.center_lng and params.radius_miles: 

134 # Approximate degrees per mile at this latitude 

135 lat_delta = params.radius_miles / 69.0 

136 lng_delta = params.radius_miles / (69.0 * abs(cos(radians(params.center_lat)))) 

137 

138 geo_join = "JOIN property_geo g ON p.id = g.property_id" 

139 conditions.append("g.min_lng >= :box_min_lng AND g.max_lng <= :box_max_lng") 

140 conditions.append("g.min_lat >= :box_min_lat AND g.max_lat <= :box_max_lat") 

141 bind_params.update({ 

142 "box_min_lat": params.center_lat - lat_delta, 

143 "box_max_lat": params.center_lat + lat_delta, 

144 "box_min_lng": params.center_lng - lng_delta, 

145 "box_max_lng": params.center_lng + lng_delta, 

146 "center_lat": params.center_lat, 

147 "center_lng": params.center_lng, 

148 "radius": params.radius_miles, 

149 }) 

150 

151 # Add Haversine distance calculation 

152 distance_select = """, 

153 (3959 * acos( 

154 cos(radians(:center_lat)) * cos(radians(p.latitude)) 

155 * cos(radians(p.longitude) - radians(:center_lng)) 

156 + sin(radians(:center_lat)) * sin(radians(p.latitude)) 

157 )) as distance_miles 

158 """ 

159 conditions.append("distance_miles <= :radius") 

160 

161 where_clause = " AND ".join(conditions) 

162 

163 # Sorting 

164 valid_sorts = ["list_date", "list_price", "bedrooms_total", "living_area", "year_built"] 

165 sort_col = params.sort_by if params.sort_by in valid_sorts else "list_date" 

166 sort_dir = "DESC" if params.sort_order.lower() == "desc" else "ASC" 

167 

168 # Count query 

169 count_sql = f""" 

170 SELECT COUNT(*) FROM properties p 

171 {geo_join} 

172 WHERE {where_clause} 

173 """ 

174 total = db.execute(text(count_sql), bind_params).scalar() or 0 

175 

176 # Calculate pagination 

177 offset = (params.page - 1) * params.page_size 

178 total_pages = (total + params.page_size - 1) // params.page_size 

179 

180 # Main query 

181 query_sql = f""" 

182 SELECT p.* {distance_select} 

183 FROM properties p 

184 {geo_join} 

185 WHERE {where_clause} 

186 ORDER BY p.{sort_col} {sort_dir} 

187 LIMIT :limit OFFSET :offset 

188 """ 

189 bind_params["limit"] = params.page_size 

190 bind_params["offset"] = offset 

191 

192 results = db.execute(text(query_sql), bind_params).mappings().all() 

193 

194 # Convert to response models 

195 items = [PropertyListItem.model_validate(dict(r)) for r in results] 

196 

197 query_time = (time.perf_counter() - start_time) * 1000 

198 

199 return PropertySearchResponse( 

200 items=items, 

201 total=total, 

202 page=params.page, 

203 page_size=params.page_size, 

204 total_pages=total_pages, 

205 query_time_ms=round(query_time, 2), 

206 search_type="geo" if geo_join else "filter", 

207 ) 

208 

209 

210@router.get("/{listing_id}", response_model=PropertyDetail) 

211async def get_property( 

212 listing_id: str, 

213 db: Session = Depends(get_db), 

214): 

215 """Get full details for a specific property.""" 

216 result = db.execute( 

217 text("SELECT * FROM properties WHERE listing_id = :id AND mlg_can_view = 1"), 

218 {"id": listing_id}, 

219 ).mappings().first() 

220 

221 if not result: 

222 raise HTTPException(status_code=404, detail="Property not found") 

223 

224 return PropertyDetail.model_validate(dict(result)) 

225 

226 

227@router.get("/{listing_id}/pdf") 

228async def get_property_pdf( 

229 listing_id: str, 

230 db: Session = Depends(get_db), 

231): 

232 """Generate PDF brochure for a property.""" 

233 # Fetch property data 

234 result = db.execute( 

235 text("SELECT * FROM properties WHERE listing_id = :id AND mlg_can_view = 1"), 

236 {"id": listing_id}, 

237 ).mappings().first() 

238 

239 if not result: 

240 raise HTTPException(status_code=404, detail="Property not found") 

241 

242 property_data = dict(result) 

243 

244 # Get site URL from environment 

245 site_url = os.environ.get("SITE_URL", "https://ei.supported.systems") 

246 

247 # Load broker/agent information from site.json 

248 site_config = load_site_config() 

249 

250 # Generate PDF with broker info 

251 pdf_bytes = generate_property_pdf(property_data, site_url, site_config) 

252 

253 # Create filename from address 

254 address_parts = [ 

255 property_data.get("street_number"), 

256 property_data.get("street_name"), 

257 property_data.get("street_suffix"), 

258 ] 

259 address = "_".join(filter(None, address_parts)) or listing_id 

260 filename = f"{address}_{listing_id}.pdf" 

261 

262 return Response( 

263 content=pdf_bytes, 

264 media_type="application/pdf", 

265 headers={ 

266 "Content-Disposition": f'inline; filename="{filename}"', 

267 "Cache-Control": "public, max-age=3600", 

268 }, 

269 ) 

270 

271 

272@router.get("/cities/list") 

273async def list_cities( 

274 state: str | None = Query(None), 

275 db: Session = Depends(get_db), 

276): 

277 """Get list of cities with property counts.""" 

278 conditions = ["mlg_can_view = 1", "standard_status = 'Active'"] 

279 params = {} 

280 

281 if state: 

282 conditions.append("state_or_province = :state") 

283 params["state"] = state 

284 

285 where = " AND ".join(conditions) 

286 

287 results = db.execute( 

288 text(f""" 

289 SELECT city, state_or_province, COUNT(*) as count 

290 FROM properties 

291 WHERE {where} 

292 GROUP BY city, state_or_province 

293 ORDER BY count DESC 

294 LIMIT 100 

295 """), 

296 params, 

297 ).mappings().all() 

298 

299 return [dict(r) for r in results] 

300 

301 

302# Import math for radius search 

303from math import cos, radians