Natuurlijk! Hieronder vind je een uitgebreide foutafhandelingsstrategie voor jouw Python-webapplicatie, gericht op network timeouts en andere potentiële fouten, inclusief best practices en voorbeeldcode. --- ## 1. Algemene Best Practices voor Foutafhandeling in Python - **Gebruik try-except blokken:** Plaats kritieke code binnen try-blokken om uitzonderingen af te vangen. - **Specifieke except-blokken:** Vang specifieke exceptions (bijv. `requests.exceptions.Timeout`) om gerichte acties te ondernemen. - **Logging:** Log fouten voor later onderzoek en debugging. - **Herhaalpogingen:** Bij tijdelijke fouten zoals timeouts, probeer de actie opnieuw met een exponential backoff. - **Gebruik van context managers:** Voor resourcebeheer en netwerksessies. - **Graceful degrade:** Geef een duidelijke foutmelding aan de gebruiker indien nodig, in plaats van een crash. --- ## 2. Specifieke Strategie voor Network Timeouts ### a. Gebruik van `requests` bibliotheek met timeouts ```python import requests import time import logging logging.basicConfig(level=logging.INFO) def fetch_data_with_retries(url, max_retries=3): retries = 0 backoff_factor = 1 # seconden while retries < max_retries: try: response = requests.get(url, timeout=5) # 5 seconden timeout response.raise_for_status() return response.json() except requests.exceptions.Timeout: retries += 1 wait_time = backoff_factor * (2 ** (retries - 1)) logging.warning(f"Timeout bij poging {retries} voor {url}. Wacht {wait_time} seconden voordat opnieuw geprobeerd.") time.sleep(wait_time) except requests.exceptions.RequestException as e: logging.error(f"Netwerkfout: {e}") break logging.error(f"Alle pogingen voor {url} zijn mislukt.") return None ``` ### b. Uitleg - **Timeout instellen:** `timeout=5` geeft aan dat elke request na 5 seconden stopt. - **Herhaalpogingen:** Bij een timeout wordt de request herhaald totdat het maximum aantal retries is bereikt. - **Exponential backoff:** Wacht tijd neemt toe na elke mislukte poging, om serverbelasting te verminderen. - **Logging:** Geeft inzicht in het proces en mogelijke problemen. --- ## 3. Foutafhandeling in Frameworks (bijv. Flask/Django) Voor webframeworks is het belangrijk om fouten op te vangen en gepaste HTTP-responses te sturen. ### Flask voorbeeld: ```python from flask import Flask, jsonify app = Flask(__name__) @app.errorhandler(Exception) def handle_exception(e): # Log de fout app.logger.error(f"Onverwachte fout: {e}") # Stuur een algemene foutmelding return jsonify(error="Er is een interne serverfout opgetreden."), 500 @app.route('/api/data') def get_data(): try: data = fetch_data_with_retries('https://api.example.com/data') if data is None: return jsonify(error="Unable to ophalen data."), 503 return jsonify(data) except Exception as e: app.logger.error(f"Fout bij /api/data: {e}") return jsonify(error="Fout bij het verwerken van de aanvraag."), 500 ``` --- ## 4. Extra Tips - **Gebruik van retries en timeouts in alle netwerkcommunicatie.** - **Geef gebruikers duidelijke foutmeldingen en instructies.** - **Implementeer monitoring en alerts voor frequent voorkomende fouten.** - **Test je foutafhandelingscode regelmatig met gesimuleerde foutscenario's.** --- Mocht je specifieke code of implementatievoorbeelden voor andere onderdelen van je applicatie willen, laat het gerust weten!

Zeker! Hier is een uitgebreide foutafhandelingsstrategie voor netwerk timeouts in Python webapplicaties met Amazon CodeWhisperer: ## 1. Basis Strategie en Best Practices ### Best Practices voor Network Timeouts: - **Expliciete timeout instellingen** voor alle externe calls - **Retry mechanisme** met exponential backoff - **Circuit breaker pattern** om overbelaste services te beschermen - **Gedetailleerde logging** voor debugging - **Gebruikersvriendelijke foutmeldingen** ## 2. Core Foutafhandelingsklasse ```python import logging import time from functools import wraps from typing import Callable, Any, Optional import requests from requests.exceptions import Timeout, ConnectionError, RequestException class NetworkErrorHandler: def __init__(self, max_retries: int = 3, base_delay: float = 1.0): self.max_retries = max_retries self.base_delay = base_delay self.logger = logging.getLogger(__name__) def exponential_backoff(self, attempt: int) -> float: """Bereken exponential backoff delay""" return self.base_delay * (2 ** attempt) def retry_on_timeout(self, func: Callable) -> Callable: """Decorator voor automatic retry op timeouts""" @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(self.max_retries + 1): try: return func(*args, **kwargs) except (Timeout, ConnectionError) as e: last_exception = e if attempt < self.max_retries: delay = self.exponential_backoff(attempt) self.logger.warning( f"Timeout bij poging {attempt + 1}/{self.max_retries + 1}. " f"Opnieuw proberen over {delay} seconden. Fout: {str(e)}" ) time.sleep(delay) else: self.logger.error( f"Alle {self.max_retries + 1} pogingen mislukt voor {func.__name__}" ) raise last_exception return wrapper ``` ## 3. HTTP Client met Robuuste Timeout Afhandeling ```python class RobustHTTPClient: def __init__(self, default_timeout: tuple = (5, 30)): self.session = requests.Session() self.default_timeout = default_timeout self.error_handler = NetworkErrorHandler() # Configureer session voor betere timeout handling self.session.headers.update({ 'User-Agent': 'MyWebApp/1.0', 'Content-Type': 'application/json' }) @NetworkErrorHandler.retry_on_timeout def get_with_timeout(self, url: str, **kwargs) -> requests.Response: """HTTP GET request met timeout handling en retry mechanisme""" timeout = kwargs.pop('timeout', self.default_timeout) try: response = self.session.get(url, timeout=timeout, **kwargs) response.raise_for_status() return response except Timeout as e: self.error_handler.logger.error(f"Timeout bij GET request naar {url}: {str(e)}") raise except requests.exceptions.HTTPError as e: self.error_handler.logger.error(f"HTTP fout {e.response.status_code} bij request naar {url}") raise except ConnectionError as e: self.error_handler.logger.error(f"Connectie fout bij request naar {url}: {str(e)}") raise def post_with_timeout(self, url: str, data: Any = None, **kwargs) -> requests.Response: """HTTP POST request met timeout handling""" return self.error_handler.retry_on_timeout( lambda: self._post_implementation(url, data, **kwargs) )() def _post_implementation(self, url: str, data: Any = None, **kwargs) -> requests.Response: timeout = kwargs.pop('timeout', self.default_timeout) response = self.session.post(url, json=data, timeout=timeout, **kwargs) response.raise_for_status() return response ``` ## 4. Circuit Breaker Pattern Implementatie ```python class CircuitBreaker: def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60): self.failure_threshold = failure_threshold self.reset_timeout = reset_timeout self.failure_count = 0 self.last_failure_time = None self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN self.logger = logging.getLogger(__name__) def can_execute(self) -> bool: """Bepaal of request uitgevoerd mag worden""" if self.state == "OPEN": if time.time() - self.last_failure_time > self.reset_timeout: self.state = "HALF_OPEN" self.logger.info("Circuit breaker naar HALF_OPEN state") else: return False return True def record_success(self): """Registreer succesvolle request""" if self.state == "HALF_OPEN": self.state = "CLOSED" self.failure_count = 0 self.last_failure_time = None def record_failure(self): """Registreer gefaalde request""" self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "OPEN" self.logger.warning(f"Circuit breaker naar OPEN state na {self.failure_count} failures") def execute(self, func: Callable, *args, **kwargs) -> Any: """Voer functie uit met circuit breaker protection""" if not self.can_execute(): raise CircuitBreakerOpenError("Service tijdelijk niet beschikbaar") try: result = func(*args, **kwargs) self.record_success() return result except Exception as e: self.record_failure() raise class CircuitBreakerOpenError(Exception): pass ``` ## 5. Gebruik in Web Application Context ```python from flask import Flask, jsonify, request import os app = Flask(__name__) # Configureer logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialiseer clients http_client = RobustHTTPClient(default_timeout=(3, 10)) external_service_circuit_breaker = CircuitBreaker() @app.route('/api/external-data', methods=['GET']) def get_external_data(): """Endpoint voor ophalen externe data met robuuste error handling""" try: # Gebruik circuit breaker voor externe service calls response = external_service_circuit_breaker.execute( http_client.get_with_timeout, 'https://api.externe-service.com/data', params=request.args ) return jsonify({ 'status': 'success', 'data': response.json() }), 200 except Timeout: logger.error("Timeout bij ophalen externe data") return jsonify({ 'status': 'error', 'message': 'Externe service reageert niet tijdig', 'code': 'SERVICE_TIMEOUT' }), 504 except ConnectionError: logger.error("Connectie fout bij externe service") return jsonify({ 'status': 'error', 'message': 'Kan geen verbinding maken met externe service', 'code': 'CONNECTION_ERROR' }), 502 except CircuitBreakerOpenError: logger.warning("Circuit breaker open - service tijdelijk niet beschikbaar") return jsonify({ 'status': 'error', 'message': 'Service is tijdelijk niet beschikbaar. Probeer het later opnieuw.', 'code': 'SERVICE_UNAVAILABLE' }), 503 except Exception as e: logger.error(f"Onverwachte fout: {str(e)}") return jsonify({ 'status': 'error', 'message': 'Interne server fout', 'code': 'INTERNAL_ERROR' }), 500 @app.errorhandler(504) def handle_gateway_timeout(error): """Global handler voor gateway timeouts""" return jsonify({ 'status': 'error', 'message': 'Backend service niet beschikbaar', 'code': 'GATEWAY_TIMEOUT' }), 504 ``` ## 6. Configuratie en Monitoring ```python # requirements.txt toevoegen """ requests>=2.28.0 flask>=2.0.0 """ # Environment configuratie class Config: # Timeout instellingen per environment HTTP_CONNECT_TIMEOUT = int(os.getenv('HTTP_CONNECT_TIMEOUT', '5')) HTTP_READ_TIMEOUT = int(os.getenv('HTTP_READ_TIMEOUT', '30')) MAX_RETRIES = int(os.getenv('MAX_RETRIES', '3')) # Circuit breaker settings CIRCUIT_BREAKER_FAILURE_THRESHOLD = int(os.getenv('CIRCUIT_BREAKER_THRESHOLD', '5')) CIRCUIT_BREAKER_RESET_TIMEOUT = int(os.getenv('CIRCUIT_BREAKER_RESET', '60')) # Monitoring en metrics def setup_metrics(): """Setup voor monitoring van network timeouts""" # Implementeer metrics collection voor: # - Aantal timeouts per service # - Circuit breaker state changes # - Response tijden # - Retry statistics pass ``` ## 7. CodeWhisperer Tips ### Commentaar voor betere suggestions: ```python # TODO: Implement exponential backoff for retry mechanism # TODO: Add circuit breaker pattern for external API calls # TODO: Create custom exception classes for network errors # TODO: Implement structured logging for timeout events ``` ## 8. Test Strategie ```python import unittest from unittest.mock import patch, MagicMock class TestNetworkErrorHandling(unittest.TestCase): def test_retry_mechanism_on_timeout(self): """Test retry gedrag bij timeouts""" client = RobustHTTPClient() with patch('requests.Session.get') as mock_get: mock_get.side_effect = Timeout("Simulated timeout") with self.assertRaises(Timeout): client.get_with_timeout('http://test.com') # Verifieer aantal retry pogingen self.assertEqual(mock_get.call_count, 4) # 1 initieel + 3 retries def test_circuit_breaker_opens_after_threshold(self): """Test circuit breaker open state na failure threshold""" breaker = CircuitBreaker(failure_threshold=2, reset_timeout=10) # Simuleer failures for _ in range(2): breaker.record_failure() self.assertEqual(breaker.state, "OPEN") self.assertFalse(breaker.can_execute()) ``` Deze strategie biedt een complete oplossing voor netwerk timeout afhandeling in je Python webapplicatie, met focus op robuustheid, gebruikerservaring en maintainability.