Hier is een basisvoorbeeld van een Python-script dat data synchroniseert tussen de twee API's, inclusief authenticatie, foutafhandeling en data-transformatie. Daarnaast leg ik uit hoe je API-verzoeken kunt optimaliseren en de synchronisatie kunt plannen. ```python import requests import time # Configuratie BRON_API_URL = "https://api.bronbedrijf.com/v1/data" DOEL_API_URL = "https://api.doelbedrijf.com/v1/update" BRON_API_KEY = "jouw_bron_api_key" # Vervang door je echte API-sleutel DOEL_API_KEY = "jouw_doel_api_key" # Vervang door je echte API-sleutel # Headers voor authenticatie headers_bron = { 'Authorization': f'Bearer {BRON_API_KEY}', 'Accept': 'application/json' } headers_doel = { 'Authorization': f'Bearer {DOEL_API_KEY}', 'Content-Type': 'application/json' } def fetch_bron_data(): try: response = requests.get(BRON_API_URL, headers=headers_bron) response.raise_for_status() data = response.json() return data except requests.exceptions.HTTPError as errh: print(f"HTTP fout bij ophalen data: {errh}") except requests.exceptions.RequestException as err: print(f"Request fout: {err}") return None def transform_data(data): # Pas deze functie aan op basis van de vereiste data-transformatie transformed = [] for item in data: # Bijvoorbeeld: hernoem velden of filter gegevens transformed_item = { 'id': item.get('id'), 'naam': item.get('naam'), 'waarde': item.get('waarde') } transformed.append(transformed_item) return transformed def update_doel(data): try: response = requests.post(DOEL_API_URL, headers=headers_doel, json=data) response.raise_for_status() print("Data succesvol geüpdatet.") except requests.exceptions.HTTPError as errh: print(f"HTTP fout bij updaten data: {errh}") except requests.exceptions.RequestException as err: print(f"Request fout: {err}") def main(): data = fetch_bron_data() if data: transformed_data = transform_data(data) update_doel(transformed_data) if __name__ == "__main__": main() ``` **Hoe API-verzoeken optimaliseren:** - **Batch-verwerking:** Als de API grote datasets ondersteunt, probeer dan data in batches te verzenden i.p.v. alles in één keer. Dit vermindert de kans op time-outs. - **Caching:** Vermijd onnodige herhaalde verzoeken door data te cachen en alleen nieuwe of gewijzigde data te synchroniseren. - **Verzoeklimieten:** Controleer de API-documentatie op rate limits en pas je verzoekfrequentie hierop aan. - **Asynchrone verzoeken:** Gebruik bijvoorbeeld `asyncio` voor parallelle verzoeken als je meerdere API's moet aanspreken. **Synchronisatie plannen:** - **Cron jobs:** Op Linux-systemen kun je het script plannen met een cron job, bijvoorbeeld elke uur: ```bash 0 * * * * /usr/bin/python3 /pad/naar/script.py ``` - **Scheduled Tasks (Windows):** Gebruik Taakplanner om het script periodiek uit te voeren. - **Automatisering tools:** Overweeg platforms zoals Apache Airflow voor meer complexe workflows. Door bovenstaande aanpak kun je een robuuste en efficiënte data-synchronisatie opzetten.

Ik help je graag met een Python script voor API-synchronisatie. Hier is een complete oplossing: ```python import requests import json import logging from datetime import datetime from typing import Dict, List, Optional import schedule import time # Configuratie BRON_API_URL = "https://api.bronbedrijf.com/v1/data" DOEL_API_URL = "https://api.doelbedrijf.com/v1/update" BRON_API_KEY = "jouw_bron_api_key" # Vervang met je eigen key DOEL_API_KEY = "jouw_doel_api_key" # Vervang met je eigen key # Logging instellen logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class APISynchronisatie: def __init__(self): self.sessie = requests.Session() self.sessie.headers.update({ 'Content-Type': 'application/json', 'User-Agent': 'API-Sync-Script/1.0' }) def haal_bron_data_op(self) -> Optional[List[Dict]]: """Haal data op van de bron API""" try: headers = {'Authorization': f'Bearer {BRON_API_KEY}'} response = self.sessie.get(BRON_API_URL, headers=headers, timeout=30) response.raise_for_status() data = response.json() logger.info(f"Successvol {len(data)} items opgehaald van bron API") return data except requests.exceptions.RequestException as e: logger.error(f"Fout bij ophalen bron data: {e}") return None except json.JSONDecodeError as e: logger.error(f"Ongeldige JSON response: {e}") return None def transformeer_data(self, data: List[Dict]) -> List[Dict]: """Transformeer data naar doel API formaat""" getransformeerde_data = [] for item in data: # Pas hier je transformatielogica aan getransformeerd_item = { 'id': item.get('id'), 'naam': item.get('name', '').title(), 'waarde': float(item.get('value', 0)), 'timestamp': datetime.now().isoformat(), 'metadata': { 'bron': 'api.bronbedrijf.com', 'sync_tijd': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } } getransformeerde_data.append(getransformeerd_item) logger.info(f"Data getransformeerd: {len(getransformeerde_data)} items") return getransformeerde_data def verstuur_naar_doel(self, data: List[Dict]) -> bool: """Verstuur data naar doel API""" try: headers = { 'Authorization': f'Bearer {DOEL_API_KEY}', 'X-API-Key': DOEL_API_KEY } # Batch verwerking voor optimalisatie batch_grootte = 50 success_count = 0 for i in range(0, len(data), batch_grootte): batch = data[i:i + batch_grootte] response = self.sessie.post( DOEL_API_URL, headers=headers, json=batch, timeout=60 ) response.raise_for_status() success_count += len(batch) logger.info(f"Batch succesvol verzonden: {len(batch)} items") logger.info(f"Totaal {success_count} items succesvol gesynchroniseerd") return True except requests.exceptions.RequestException as e: logger.error(f"Fout bij verzenden naar doel API: {e}") if hasattr(e, 'response') and e.response is not None: logger.error(f"Response status: {e.response.status_code}") logger.error(f"Response body: {e.response.text}") return False def synchroniseer(self): """Hoofdfunctie voor synchronisatie""" logger.info("Start synchronisatie...") # Haal data op bron_data = self.haal_bron_data_op() if not bron_data: logger.error("Kon geen data ophalen van bron API") return False # Transformeer data getransformeerde_data = self.transformeer_data(bron_data) # Verstuur naar doel success = self.verstuur_naar_doel(getransformeerde_data) if success: logger.info("Synchronisatie voltooid") else: logger.error("Synchronisatie mislukt") return success # Optimalisatie technieken def optimalisatie_tips(): """Tips voor API-verzoek optimalisatie""" tips = [ "1. Gebruik paginering als de API het ondersteunt", "2. Implementeer batch verwerking (zoals in het script)", "3. Gebruik compression (gzip) voor grote payloads", "4. Cache responses waar mogelijk", "5. Gebruik conditional requests (ETag, Last-Modified)", "6. Paralleliseer requests met threading voor grote datasets", "7. Implementeer retry mechanisme met exponential backoff", "8. Monitor en optimaliseer payload grootte" ] return tips # Planning en uitvoering def plan_synchronisatie(): """Plan de synchronisatie met schedule library""" sync = APISynchronisatie() # Plan synchronisatie (voorbeeld: elk uur) schedule.every(1).hours.do(sync.synchroniseer) # Eerste sync direct uitvoeren sync.synchroniseer() logger.info("Synchronisatie gepland - draait elk uur") # Oneindige loop voor planning while True: schedule.run_pending() time.sleep(60) # Check elke minuut if __name__ == "__main__": # Toon optimalisatie tips print("API Optimalisatie Tips:") for tip in optimalisatie_tips(): print(f" {tip}") print("\nStart synchronisatie...") # Voor eenmalige sync: # sync = APISynchronisatie() # sync.synchroniseer() # Voor geplande sync (uncomment onderstaande regel): # plan_synchronisatie() ``` ## Installatie vereisten: ```bash pip install requests schedule ``` ## Belangrijke optimalisatietechnieken: 1. **Batch verwerking**: Verzend data in batches i.p.v. per item 2. **Paginering**: Gebruik `?page=1&limit=100` parameters als beschikbaar 3. **Parallel requests**: Gebruik `concurrent.futures` voor parallelle verwerking 4. **Caching**: Sla vaak opgevraagde data lokaal op 5. **Compressie**: Gebruik gzip compression voor grote payloads ## Synchronisatie plannen: 1. **Windows Task Scheduler** of **cron jobs** voor OS-level planning 2. **Python schedule library** (zoals in het voorbeeld) 3. **Celery** met **Redis** voor geavanceerde job queueing 4. **AWS Lambda** + **CloudWatch Events** voor serverless planning ## Beveiligingstips: - Sla API keys op in environment variables - Gebruik HTTPS voor alle verzoeken - Implementeer rate limiting - Log gevoelige data niet - Gebruik retry mechanismes met backoff Pas de transformatielogica aan op basis van je specifieke API structuren!