Webhook Receiver Implementatie

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import json
from typing import Dict, Any

app = FastAPI()

class WebhookProcessor:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
        self.event_handlers = {}
        
    def verify_signature(self, payload: bytes, signature: str) -> bool:
        """Verify webhook signature voor security"""
        expected = hmac.new(
            self.secret_key.encode(),
            payload,
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(expected, signature)
    
    def register_handler(self, event_type: str, handler):
        """Registreer handler voor specifiek event type"""
        self.event_handlers[event_type] = handler
    
    async def process_event(self, event: Dict[str, Any]):
        """Verwerk webhook event"""
        event_type = event.get('type')
        if event_type in self.event_handlers:
            await self.event_handlers[event_type](event)
        else:
            print(f"Geen handler voor event type: {event_type}")

processor = WebhookProcessor(secret_key="jouw-secret-key")

@app.post("/webhook")
async def receive_webhook(request: Request):
    # Lees payload en signature
    payload = await request.body()
    signature = request.headers.get('X-Webhook-Signature')
    
    # Verifieer signature
    if not processor.verify_signature(payload, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Parse en verwerk event
    event = json.loads(payload.decode())
    await processor.process_event(event)
    
    return {"status": "success"}

# Registreer event handlers
@processor.register_handler("order.created")
async def handle_order_created(event: Dict[str, Any]):
    """Verwerk nieuwe order"""
    order_data = event['data']
    # Sla op in database of queue voor verdere verwerking
    print(f"Nieuwe order ontvangen: {order_data['id']}")

@processor.register_handler("customer.updated")
async def handle_customer_update(event: Dict[str, Any]):
    """Verwerk klant update"""
    customer_data = event['data']
    # Update customer record in data warehouse
    print(f"Klant bijgewerkt: {customer_data['email']}")

Webhook Best Practices

• Signature Verification: Altijd implementeren voor security
• Idempotency Keys: Voorkom duplicate processing
• Retry Logic: Implementeer dead letter queue voor failed events
• Event Schema Versioning: Behandel backward compatibility
• Async Processing: Verwerk events asynchroon om timeouts te voorkomen

WebSocket Streaming

import asyncio
import websockets
import json
import pandas as pd
from datetime import datetime
from collections import deque

class WebSocketStreamer:
    def __init__(self, uri: str, max_buffer: int = 10000):
        self.uri = uri
        self.buffer = deque(maxlen=max_buffer)
        self.running = False
        
    async def connect_and_stream(self):
        """Maak verbinding en stream real-time data"""
        self.running = True
        
        async with websockets.connect(self.uri) as websocket:
            # Subscribe naar events
            await websocket.send(json.dumps({
                "action": "subscribe",
                "channels": ["trades", "orderbook"]
            }))
            
            # Ontvang en verwerk messages
            async for message in websocket:
                if not self.running:
                    break
                    
                data = json.loads(message)
                await self.process_message(data)
    
    async def process_message(self, data: dict):
        """Verwerk incoming message"""
        event_type = data.get('type')
        
        if event_type == 'trade':
            trade_data = {
                'timestamp': datetime.now(),
                'symbol': data['symbol'],
                'price': float(data['price']),
                'quantity': float(data['quantity']),
                'side': data['side']
            }
            self.buffer.append(trade_data)
            
            # Batch verwerking elke 100 trades
            if len(self.buffer) >= 100:
                await self.process_batch()
                
        elif event_type == 'orderbook_update':
            # Update orderbook state
            pass
    
    async def process_batch(self):
        """Verwerk batch van trades"""
        if not self.buffer:
            return
            
        batch = list(self.buffer)
        df = pd.DataFrame(batch)
        
        # Aggregaties berekenen
        aggregations = df.groupby('symbol').agg({
            'price': ['mean', 'min', 'max'],
            'quantity': 'sum'
        }).round(4)
        
        # Sla op naar database of publish naar message queue
        print(f"Verwerkt batch van {len(batch)} trades")
        print(aggregations)
        
        # Clear buffer na verwerking
        self.buffer.clear()
    
    def stop(self):
        """Stop de stream"""
        self.running = False

Streaming Best Practices

• Connection Management: Implementeer reconnection logic
• Backpressure Handling: Buffer management bij hoge volumes
• State Management: Bewaar connection state voor recovery
• Monitoring: Track message rates en latency
• Batch Processing: Verwerk in batches voor efficiency