Ontwikkel een webhook implementatie met geavanceerde payload validatie en filtering
description
Met deze prompt kan de gebruiker een webhook systeem ontwerpen dat alleen valide en relevante data verwerkt, wat de betrouwbaarheid en prestaties verbetert. Het voorkomt dat ongeldige of onnodige payloads de applicatie belasten en helpt bij het monitoren van de datastroom, wat alternatieven zonder filtering en validatie niet bieden.
prompt
Help me een webhook implementatie te ontwerpen waarbij ik geavanceerde payload validatie en filtering kan toepassen. Geef advies over het opzetten van regels voor het valideren van inkomende payloads op basis van <payloa ...
try_prompt
generate_helper
disclaimerOnPageApi
tags
ai_answers
provider: openai
model: gpt-4.1-nano
Natuurlijk! Hier is een gedetailleerd advies voor het ontwerpen van een webhook-implementatie met geavanceerde payload validatie en filtering, speciaal voor jouw situatie:
1. Ontvangst van Webhook
- Endpoint: Gebruik jouw URL https://api.mijnapplicatie.nl/webhook
- Method: POST
- Content-Type: application/json
2. Basis Validatie van Payload
- Controleer of de payload geldig JSON is.
- Valideer of de JSON de vereiste structuur bevat, namelijk:
```json
{
"event_type": "string",
"data": "object"
}
```
- Bijvoorbeeld in pseudocode:
```python
payload = json.loads(request.body)
if not isinstance(payload, dict):
# Ongeldige payload
return 400
if "event_type" not in payload or "data" not in payload:
# Ontbrekende velden
return 400
if not isinstance(payload["event_type"], str):
# Ongeldige type voor event_type
return 400
if not isinstance(payload["data"], dict):
# Ongeldige data
return 400
```
3. Filtering op event_type
- Alleen verwerken van webhook events waar `event_type` gelijk is aan `"order_created"`:
```python
if payload["event_type"] != "order_created":
# Niet relevant, log en negeer
log_event("Filtered event_type: " + payload["event_type"])
return 204 # Geen inhoud om te verwerken
```
4. Geavanceerde Payload Validatie
- Validatie van data-velden (bijvoorbeeld verwacht veld `order_id`, `amount`, etc.):
```python
required_fields = ["order_id", "amount"]
for field in required_fields:
if field not in payload["data"]:
log_event(f"Ontbrekend veld {field} in data")
return 400
```
- Controleer dat `amount` bijvoorbeeld een positief getal is:
```python
if not isinstance(payload["data"]["amount"], (int, float)) or payload["data"]["amount"] <= 0:
log_event("Ongeldige waarde voor amount")
return 400
```
5. Afwijzen van Ongeldige of Ongewenste Payloads
- Bij onjuiste structuur, ontbrekende velden of ongeldige data:
- Log de fout met details voor monitoring:
```python
log_event(f"Ongeldige payload: {payload}")
```
- Geef een passende HTTP-statuscode terug:
- 400 Bad Request voor ongeldige payloads.
- 403 Forbidden of 422 Unprocessable Entity kan ook passend zijn afhankelijk van je beleid.
6. Logging en Monitoring
- Log alle ontvangen payloads die worden afgewezen of gefilterd.
- Gebruik een gestructureerde loggingbibliotheek of service (bijvoorbeeld ELK-stack, Datadog).
- Log belangrijke details zoals timestamp, event_type, data, en foutmeldingen.
- Bijvoorbeeld:
```python
def log_event(message):
# Implementatie afhankelijk van je logging setup
print(f"[{datetime.now()}] {message}")
```
7. Beveiliging
- Controleer of de request afkomstig is van vertrouwde bronnen (bijvoorbeeld via een secret token of signature).
- Bijvoorbeeld door een header `X-Hub-Signature` te controleren.
8. Samenvatting van de Workflow
- Ontvangst van payload
- Validatie van JSON structuur
- Filtering op event_type == "order_created"
- Geavanceerde validatie van data velden
- Afwijzen en loggen bij fouten
- Verwerking van geldige payloads
Door deze stappen te volgen, zorg je voor een robuuste webhook-implementatie met goede validatie, filtering, logging en monitoring.
provider: deepseek
model: deepseek-chat
Zeker, hier is een gestructureerd ontwerp voor je webhook implementatie met geavanceerde validatie en filtering:
## 1. Algemene Architectuur
```python
# Webhook ontvangst endpoint (bijv. Flask)
@app.route('/webhook', methods=['POST'])
def handle_webhook():
try:
# Stap 1: Basis validatie
if not request.is_json:
return jsonify({"error": "Alleen JSON payloads toegestaan"}), 400
payload = request.get_json()
# Stap 2: Schema validatie
validation_result = validate_payload_schema(payload)
if not validation_result["valid"]:
log_invalid_payload(payload, validation_result["errors"])
return jsonify({"error": "Ongeldige payload structuur"}), 400
# Stap 3: Event filtering
if payload["event_type"] == "order_created":
# Stap 4: Geavanceerde validatie
if not validate_order_data(payload["data"]):
log_invalid_payload(payload, "Ongeldige order data")
return jsonify({"error": "Ongeldige order informatie"}), 400
# Verwerk geldige payload
process_order_created(payload)
log_successful_processing(payload)
return jsonify({"status": "success"}), 200
except Exception as e:
log_processing_error(payload, str(e))
return jsonify({"error": "Interne serverfout"}), 500
```
## 2. Geavanceerde Validatie Regels
### Schema Validatie (JSON Schema)
```python
import jsonschema
WEBHOOK_SCHEMA = {
"type": "object",
"required": ["event_type", "data"],
"properties": {
"event_type": {
"type": "string",
"enum": ["order_created", "order_updated", "order_cancelled"]
},
"data": {
"type": "object",
"required": ["order_id", "customer_id", "amount"],
"properties": {
"order_id": {"type": "string", "pattern": "^ORD_\\d+$"},
"customer_id": {"type": "string"},
"amount": {"type": "number", "minimum": 0},
"currency": {"type": "string", "pattern": "^[A-Z]{3}$"},
"items": {
"type": "array",
"items": {
"type": "object",
"required": ["product_id", "quantity"],
"properties": {
"product_id": {"type": "string"},
"quantity": {"type": "integer", "minimum": 1}
}
}
}
}
}
}
}
def validate_payload_schema(payload):
try:
jsonschema.validate(instance=payload, schema=WEBHOOK_SCHEMA)
return {"valid": True}
except jsonschema.ValidationError as e:
return {"valid": False, "errors": str(e)}
```
### Aangepaste Business Validatie
```python
def validate_order_data(order_data):
# Controleer verplichte velden
required_fields = ["order_id", "customer_id", "amount"]
if not all(field in order_data for field in required_fields):
return False
# Valideer bedrag
if order_data["amount"] <= 0:
return False
# Valideer currency indien aanwezig
if "currency" in order_data and len(order_data["currency"]) != 3:
return False
return True
```
## 3. Filtering Mechanismen
### Event Type Filter
```python
EVENT_FILTERS = {
"order_created": {
"required": True,
"handler": process_order_created,
"validation_rules": validate_order_data
},
"order_updated": {
"required": False,
"handler": process_order_updated
}
}
def handle_filtered_event(payload):
event_type = payload["event_type"]
if event_type not in EVENT_FILTERS:
log_unknown_event_type(payload)
return False
filter_config = EVENT_FILTERS[event_type]
# Extra validatie voor specifieke events
if "validation_rules" in filter_config:
if not filter_config["validation_rules"](payload["data"]):
return False
# Voer event handler uit
filter_config["handler"](payload)
return True
```
## 4. Beveiliging en Afwijzing
### Rate Limiting
```python
from flask_limiter import Limiter
limiter = Limiter(app, key_func=get_remote_address)
@app.route('/webhook', methods=['POST'])
@limiter.limit("100 per hour")
def handle_webhook():
# ... implementatie
```
### IP Whitelisting (optioneel)
```python
ALLOWED_IPS = ["192.168.1.0/24", "10.0.0.1"]
def check_ip_whitelist():
client_ip = request.remote_addr
for allowed_ip in ALLOWED_IPS:
if ip_address(client_ip) in ip_network(allowed_ip):
return True
return False
```
## 5. Logging en Monitoring
### Uitgebreide Logging
```python
import logging
from datetime import datetime
logging.basicConfig(filename='webhook.log', level=logging.INFO)
def log_invalid_payload(payload, reason):
logging.warning(f"{datetime.now()} - INVALID: {reason} - Payload: {json.dumps(payload)}")
def log_successful_processing(payload):
logging.info(f"{datetime.now()} - SUCCESS: {payload['event_type']} - Order: {payload['data']['order_id']}")
def log_processing_error(payload, error):
logging.error(f"{datetime.now()} - ERROR: {error} - Payload: {json.dumps(payload)}")
```
### Monitoring Metrics
```python
from prometheus_client import Counter, Histogram
WEBHOOK_REQUESTS = Counter('webhook_requests_total', 'Total webhook requests', ['event_type', 'status'])
PROCESSING_TIME = Histogram('webhook_processing_seconds', 'Time spent processing webhooks')
@PROCESSING_TIME.time()
def process_webhook(payload):
# Verwerkingslogica
WEBHOOK_REQUESTS.labels(
event_type=payload['event_type'],
status='success'
).inc()
```
## 6. Response Handling
### Gestandaardiseerde Responses
```python
def create_response(success, message=None, details=None):
response = {
"status": "success" if success else "error",
"timestamp": datetime.now().isoformat(),
"message": message
}
if details:
response["details"] = details
return response
```
## 7. Implementatie Tips
1. **Gebruik middleware** voor algemene validatie
2. **Implementeer retry logic** voor tijdelijke fouten
3. **Gebruik environment variables** voor configuratie
4. **Test uitgebreid** met verschillende payload scenarios
5. **Monitor performance** met metrische gegevens
6. **Houd security headers** in acht (CORS, CSRF protection)
Dit ontwerp biedt een robuuste basis voor geavanceerde webhook verwerking met uitgebreide validatie, filtering en monitoring mogelijkheden.