Coverage for src / idx_api / routers / leads.py: 47%

1"""Leads API endpoints for contact form submissions.""" 

2 

3from fastapi import APIRouter, Depends, HTTPException, Request 

4from pydantic import BaseModel, EmailStr 

5from sqlalchemy import select 

6from sqlalchemy.orm import Session 

7 

8from idx_api.database import get_db 

9from idx_api.models.brokerage import Brokerage 

10from idx_api.models.brokerage_domain import BrokerageDomain 

11from idx_api.models.lead import Lead 

12 

13router = APIRouter() 

14 

15 

16# ===== Request/Response Models ===== 

17 

18 

19class ContactFormRequest(BaseModel): 

20 """Contact form submission from frontend.""" 

21 

22 firstName: str 

23 lastName: str 

24 email: EmailStr 

25 phone: str | None = None 

26 subject: str 

27 message: str 

28 # Optional UTM tracking 

29 utm_source: str | None = None 

30 utm_medium: str | None = None 

31 utm_campaign: str | None = None 

32 # Page they submitted from 

33 source_page: str | None = None 

34 

35 

36class ContactFormResponse(BaseModel): 

37 """Response after successful contact form submission.""" 

38 

39 success: bool 

40 message: str 

41 lead_id: int | None = None 

42 

43 

44# ===== Helper Functions ===== 

45 

46 

47def _get_brokerage_from_domain(domain: str, db: Session) -> Brokerage | None: 

48 """Look up brokerage by domain (same logic as site-config).""" 

49 domain_record = db.scalar( 

50 select(BrokerageDomain) 

51 .where(BrokerageDomain.domain == domain) 

52 .order_by(BrokerageDomain.is_verified.desc()) 

53 ) 

54 

55 if domain_record and domain_record.brokerage: 

56 if domain_record.brokerage.disabled_at is None: 

57 return domain_record.brokerage 

58 

59 return None 

60 

61 

62# ===== Public Endpoints ===== 

63 

64 

65@router.post("/contact", response_model=ContactFormResponse) 

66async def submit_contact_form( 

67 request: Request, 

68 form_data: ContactFormRequest, 

69 db: Session = Depends(get_db), 

70): 

71 """ 

72 Submit contact form and create a lead. 

73 

74 The brokerage is automatically determined from the request domain 

75 (Host header), ensuring leads go to the correct tenant. 

76 

77 This endpoint is public (no auth required) but rate-limited. 

78 """ 

79 # Get domain from request (same logic as site-config) 

80 host = request.headers.get("x-forwarded-host", "").lower() 

81 if not host: 

82 host = request.headers.get("host", "").lower() 

83 if ":" in host: 

84 host = host.split(":")[0] 

85 

86 if not host: 

87 raise HTTPException(status_code=400, detail="Missing Host header") 

88 

89 # Look up brokerage by domain 

90 brokerage = _get_brokerage_from_domain(host, db) 

91 

92 if not brokerage: 

93 raise HTTPException( 

94 status_code=404, 

95 detail=f"Domain '{host}' is not configured. Unable to process lead." 

96 ) 

97 

98 # Create the lead 

99 lead = Lead( 

100 first_name=form_data.firstName, 

101 last_name=form_data.lastName, 

102 email=form_data.email, 

103 phone=form_data.phone, 

104 subject=form_data.subject, 

105 message=form_data.message, 

106 source_domain=host, 

107 source_page=form_data.source_page, 

108 utm_source=form_data.utm_source, 

109 utm_medium=form_data.utm_medium, 

110 utm_campaign=form_data.utm_campaign, 

111 brokerage_id=brokerage.id, 

112 status="new", 

113 ) 

114 

115 db.add(lead) 

116 db.commit() 

117 db.refresh(lead) 

118 

119 # TODO: Send email notification to brokerage 

120 # This would be a good place to integrate with an email service 

121 # like SendGrid, Postmark, or AWS SES 

122 

123 return ContactFormResponse( 

124 success=True, 

125 message="Thank you for your message! We'll get back to you within 24 hours.", 

126 lead_id=lead.id, 

127 ) 

128 

129 

130@router.get("/leads/count") 

131async def get_leads_count( 

132 request: Request, 

133 db: Session = Depends(get_db), 

134): 

135 """ 

136 Get count of leads for the current domain's brokerage. 

137 

138 Useful for admin dashboard widgets. 

139 """ 

140 # Get domain from request 

141 host = request.headers.get("x-forwarded-host", "").lower() 

142 if not host: 

143 host = request.headers.get("host", "").lower() 

144 if ":" in host: 

145 host = host.split(":")[0] 

146 

147 brokerage = _get_brokerage_from_domain(host, db) 

148 

149 if not brokerage: 

150 raise HTTPException(status_code=404, detail="Brokerage not found") 

151 

152 # Count leads by status 

153 from sqlalchemy import func 

154 

155 counts = db.execute( 

156 select(Lead.status, func.count(Lead.id)) 

157 .where(Lead.brokerage_id == brokerage.id) 

158 .group_by(Lead.status) 

159 ).all() 

160 

161 result = {status: count for status, count in counts} 

162 result["total"] = sum(result.values()) 

163 

164 return result