Spaces:
Runtime error
Runtime error
File size: 5,393 Bytes
17847d4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """
Webhook service for form submission notifications
"""
from sqlalchemy.orm import Session
from typing import Dict, Any, Optional
import hmac
import hashlib
import logging
import asyncio
try:
import httpx
except ImportError:
httpx = None
from ..models import WebhookConfig, Form
logger = logging.getLogger(__name__)
class WebhookService:
"""Service for sending webhooks"""
def __init__(self):
self.max_retries = 3
self.retry_delays = [1, 5, 30] # seconds
self.timeout = 30 # seconds
async def send_webhook(
self,
db: Session,
form_id: int,
event_type: str,
payload: Dict[str, Any]
):
"""
Send webhook to configured URL.
Args:
db: Database session
form_id: ID of the form
event_type: Type of event (e.g., 'form_submitted_complete')
payload: Data to send
"""
if not httpx:
logger.error("httpx not installed, cannot send webhooks")
return
try:
# Get webhook configuration
webhook_config = db.query(WebhookConfig).filter(
WebhookConfig.form_id == form_id,
WebhookConfig.is_active == True
).first()
if not webhook_config:
logger.debug(f"No webhook configured for form {form_id}")
return
# Check if this event type is configured
if event_type not in webhook_config.events:
logger.debug(f"Event {event_type} not configured for form {form_id}")
return
# Sign payload
signature = self._sign_payload(payload, webhook_config.secret)
# Prepare headers
headers = {
"Content-Type": "application/json",
"X-AutoForm-Signature": signature,
"X-AutoForm-Event": event_type,
"User-Agent": "AutoForm-Webhook/1.0"
}
# Send with retries
await self._send_with_retries(
webhook_config.webhook_url,
payload,
headers
)
logger.info(f"Webhook sent successfully for form {form_id}, event {event_type}")
except Exception as e:
logger.error(f"Failed to send webhook for form {form_id}: {e}")
async def _send_with_retries(
self,
url: str,
payload: Dict[str, Any],
headers: Dict[str, str]
):
"""
Send webhook with retry logic.
Args:
url: Webhook URL
payload: Data to send
headers: HTTP headers
"""
async with httpx.AsyncClient(timeout=self.timeout) as client:
for attempt in range(self.max_retries):
try:
response = await client.post(
url,
json=payload,
headers=headers
)
# Check if successful
if 200 <= response.status_code < 300:
logger.info(f"Webhook delivered successfully to {url}")
return
else:
logger.warning(
f"Webhook attempt {attempt + 1} failed with status {response.status_code}: {response.text}"
)
except Exception as e:
logger.warning(f"Webhook attempt {attempt + 1} failed: {e}")
# Wait before retry (except on last attempt)
if attempt < self.max_retries - 1:
delay = self.retry_delays[attempt]
logger.info(f"Retrying in {delay} seconds...")
await asyncio.sleep(delay)
# All retries failed
logger.error(f"Webhook delivery failed after {self.max_retries} attempts to {url}")
def _sign_payload(self, payload: Dict[str, Any], secret: str) -> str:
"""
Sign payload with HMAC-SHA256.
Args:
payload: Data to sign
secret: Secret key
Returns:
Hexadecimal signature string
"""
import json
# Serialize payload
payload_str = json.dumps(payload, sort_keys=True)
# Generate HMAC
signature = hmac.new(
secret.encode(),
payload_str.encode(),
hashlib.sha256
).hexdigest()
return signature
def verify_signature(
self,
payload: Dict[str, Any],
signature: str,
secret: str
) -> bool:
"""
Verify webhook signature.
Args:
payload: Payload that was signed
signature: Signature to verify
secret: Secret key
Returns:
True if signature is valid
"""
expected_signature = self._sign_payload(payload, secret)
return hmac.compare_digest(signature, expected_signature)
# Singleton instance
webhook_service = WebhookService()
|