91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""Public survey submission endpoint. No authentication required."""
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.email import EmailService
|
|
from app.core.rate_limit import limiter
|
|
from app.models.survey_invite import SurveyInvite
|
|
from app.models.survey_response import SurveyResponse
|
|
from app.schemas.survey import SurveyInviteStatus, SurveySubmission, SurveySubmissionResponse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["survey"])
|
|
|
|
|
|
@router.get("/survey/invite/{token}", response_model=SurveyInviteStatus)
|
|
async def check_invite_status(
|
|
token: str,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Check if a survey invite token is valid and its status."""
|
|
result = await db.execute(
|
|
select(SurveyInvite).where(SurveyInvite.token == token)
|
|
)
|
|
invite = result.scalar_one_or_none()
|
|
if not invite:
|
|
raise HTTPException(status_code=404, detail="Invalid invite token")
|
|
return SurveyInviteStatus(name=invite.recipient_name, status=invite.status)
|
|
|
|
|
|
@router.post("/survey/submit", response_model=SurveySubmissionResponse)
|
|
@limiter.limit("3/hour")
|
|
async def submit_survey(
|
|
request: Request,
|
|
data: SurveySubmission,
|
|
db: Annotated[AsyncSession, Depends(get_db)],
|
|
):
|
|
"""Accept a public survey submission. No auth required.
|
|
|
|
Rate limited to 3/hour per IP to prevent spam.
|
|
Saves to DB and sends notification email to configured address.
|
|
"""
|
|
ip = request.client.host if request.client else None
|
|
ua = request.headers.get("user-agent", "")
|
|
|
|
invite = None
|
|
if data.token:
|
|
result = await db.execute(
|
|
select(SurveyInvite).where(SurveyInvite.token == data.token)
|
|
)
|
|
invite = result.scalar_one_or_none()
|
|
if invite and invite.status == "completed":
|
|
raise HTTPException(status_code=409, detail="This survey has already been submitted")
|
|
|
|
response = SurveyResponse(
|
|
respondent_name=data.respondent_name or (invite.recipient_name if invite else None),
|
|
responses=data.responses,
|
|
ip_address=ip,
|
|
user_agent=ua,
|
|
invite_id=invite.id if invite else None,
|
|
)
|
|
db.add(response)
|
|
|
|
if invite:
|
|
invite.status = "completed"
|
|
invite.completed_at = datetime.now(timezone.utc)
|
|
|
|
await db.flush()
|
|
|
|
try:
|
|
if settings.FEEDBACK_EMAIL:
|
|
await EmailService.send_survey_notification_email(
|
|
to_email=settings.FEEDBACK_EMAIL,
|
|
respondent_name=response.respondent_name,
|
|
responses=data.responses,
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to send survey notification email")
|
|
|
|
await db.commit()
|
|
|
|
return SurveySubmissionResponse(id=str(response.id))
|