Intégrer une API Externe avec Python dans Grist : Guide Pratique
Introduction
Grist avec Python devient une plateforme d'intégration puissante. En connectant des API externes, vous pouvez synchroniser des données en temps réel, automatiser des workflows et enrichir vos spreadsheets avec des données du monde entier.
Prérequis
- Compréhension des requêtes HTTP (GET, POST, PUT, DELETE)
- Connaissance de base de JSON
- Accès aux clés API des services que vous souhaitez intégrer
Les Bases des Requêtes API
1. Requête GET Simple
import urllib.request
import json
def fetch_data_from_api(url, headers=None):
"""
Effectue une requête GET vers une API
"""
if headers is None:
headers = {}
try:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=30) as response:
data = json.loads(response.read().decode('utf-8'))
return data
except Exception as e:
return {'error': str(e)}
# Exemple : Récupérer le taux de change
# =fetch_data_from_api('https://api.exchangerate-api.com/v4/latest/EUR')2. Requête POST avec Données
import urllib.request
import json
def post_data_to_api(url, data, headers=None):
"""
Envoie des données vers une API via POST
"""
if headers is None:
headers = {'Content-Type': 'application/json'}
try:
json_data = json.dumps(data).encode('utf-8')
req = urllib.request.Request(url, data=json_data,
headers=headers, method='POST')
with urllib.request.urlopen(req, timeout=30) as response:
return json.loads(response.read().decode('utf-8'))
except Exception as e:
return {'error': str(e)}Intégrations Pratiques
1. Taux de Change en Temps Réel
def get_exchange_rate(base_currency, target_currency):
"""
Récupère le taux de change actuel
"""
url = f"https://api.exchangerate-api.com/v4/latest/{base_currency}"
try:
data = fetch_data_from_api(url)
if 'rates' in data and target_currency in data['rates']:
return {
'rate': data['rates'][target_currency],
'date': data['date'],
'base': base_currency,
'target': target_currency
}
return {'error': 'Devise non trouvée'}
except Exception as e:
return {'error': str(e)}
# Conversion de montants
def convert_currency(amount, from_currency, to_currency):
"""
Convertit un montant d'une devise à une autre
"""
rate_data = get_exchange_rate(from_currency, to_currency)
if 'error' in rate_data:
return rate_data
converted = amount * rate_data['rate']
return {
'original_amount': amount,
'original_currency': from_currency,
'converted_amount': round(converted, 2),
'target_currency': to_currency,
'rate_used': rate_data['rate']
}
# Dans Grist
# =convert_currency($Amount, $Currency, 'EUR')2. Géocodage d'Adresses
def geocode_address(address, api_key=None):
"""
Convertit une adresse en coordonnées GPS
Utilise l'API Nominatim (OpenStreetMap)
"""
import urllib.parse
encoded_address = urllib.parse.quote(address)
url = f"https://nominatim.openstreetmap.org/search?q={encoded_address}&format=json&limit=1"
headers = {'User-Agent': 'GristApp/1.0'}
try:
data = fetch_data_from_api(url, headers)
if data and len(data) > 0:
return {
'latitude': float(data[0]['lat']),
'longitude': float(data[0]['lon']),
'display_name': data[0]['display_name'],
'osm_type': data[0]['osm_type']
}
return {'error': 'Adresse non trouvée'}
except Exception as e:
return {'error': str(e)}
# Géocodage inverse
def reverse_geocode(lat, lon):
"""
Convertit des coordonnées GPS en adresse
"""
url = f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json"
headers = {'User-Agent': 'GristApp/1.0'}
try:
data = fetch_data_from_api(url, headers)
if 'display_name' in data:
return {
'address': data['display_name'],
'city': data.get('address', {}).get('city'),
'country': data.get('address', {}).get('country'),
'postcode': data.get('address', {}).get('postcode')
}
return {'error': 'Coordonnées invalides'}
except Exception as e:
return {'error': str(e)}3. Météo en Temps Réel
def get_weather_data(lat, lon, api_key):
"""
Récupère les données météorologiques via OpenWeatherMap
"""
url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={api_key}&units=metric&lang=fr"
try:
data = fetch_data_from_api(url)
if 'main' in data:
return {
'temperature': data['main']['temp'],
'feels_like': data['main']['feels_like'],
'humidity': data['main']['humidity'],
'pressure': data['main']['pressure'],
'description': data['weather'][0]['description'],
'wind_speed': data['wind']['speed'],
'city': data['name'],
'country': data['sys']['country']
}
return {'error': data.get('message', 'Erreur inconnue')}
except Exception as e:
return {'error': str(e)}
# Prévisions sur 5 jours
def get_weather_forecast(lat, lon, api_key):
"""
Récupère les prévisions météo sur 5 jours
"""
url = f"https://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lon}&appid={api_key}&units=metric&lang=fr"
try:
data = fetch_data_from_api(url)
if 'list' in data:
forecasts = []
for item in data['list'][:5]: # 5 premières prévisions
forecasts.append({
'datetime': item['dt_txt'],
'temperature': item['main']['temp'],
'description': item['weather'][0]['description'],
'humidity': item['main']['humidity']
})
return forecasts
return {'error': 'Données non disponibles'}
except Exception as e:
return {'error': str(e)}Intégrations Business
1. Vérification d'Email (ZeroBounce)
def verify_email(email, api_key):
"""
Vérifie la validité d'une adresse email
"""
url = f"https://api.zerobounce.net/v2/validate?api_key={api_key}&email={email}"
try:
data = fetch_data_from_api(url)
return {
'email': email,
'status': data.get('status'), # valid, invalid, catch-all, unknown, spamtrap, abuse, do_not_mail
'sub_status': data.get('sub_status'),
'domain': data.get('domain'),
'did_you_mean': data.get('did_you_mean'),
'is_valid': data.get('status') == 'valid'
}
except Exception as e:
return {'error': str(e), 'email': email}2. Enrichissement de Données (Clearbit)
def enrich_company_data(domain, api_key):
"""
Enrichit les données d'une entreprise via Clearbit
"""
url = f"https://company.clearbit.com/v2/companies/find?domain={domain}"
headers = {'Authorization': f'Bearer {api_key}'}
try:
data = fetch_data_from_api(url, headers)
if 'id' in data:
return {
'name': data.get('name'),
'domain': data.get('domain'),
'industry': data.get('category', {}).get('industry'),
'employees': data.get('metrics', {}).get('employees'),
'annual_revenue': data.get('metrics', {}).get('annualRevenue'),
'city': data.get('geo', {}).get('city'),
'country': data.get('geo', {}).get('country'),
'linkedin': data.get('linkedin', {}).get('handle'),
'twitter': data.get('twitter', {}).get('handle'),
'logo': data.get('logo'),
'email': data.get('site', {}).get('emailAddresses', [None])[0]
}
return {'error': 'Entreprise non trouvée', 'domain': domain}
except Exception as e:
return {'error': str(e), 'domain': domain}3. Notifications Slack
def send_slack_notification(webhook_url, message, channel=None, username='Grist Bot'):
"""
Envoie une notification vers Slack
"""
payload = {
'text': message,
'username': username,
'icon_emoji': ':chart_with_upwards_trend:'
}
if channel:
payload['channel'] = channel
try:
result = post_data_to_api(webhook_url, payload)
return {'success': True, 'response': result}
except Exception as e:
return {'success': False, 'error': str(e)}
# Notification formatée avec blocks
def send_slack_rich_notification(webhook_url, title, fields, color='#36a64f'):
"""
Envoie une notification riche vers Slack avec formatting
"""
attachments = [{
'color': color,
'title': title,
'fields': [
{'title': field['title'], 'value': field['value'], 'short': field.get('short', True)}
for field in fields
],
'footer': 'Grist Expert',
'ts': int(datetime.now().timestamp())
}]
payload = {
'attachments': attachments,
'username': 'Grist Bot'
}
try:
result = post_data_to_api(webhook_url, payload)
return {'success': True}
except Exception as e:
return {'success': False, 'error': str(e)}Gestion des Erreurs et Retry
1. Système de Retry Intelligent
import time
def api_call_with_retry(func, max_retries=3, delay=1):
"""
Effectue un appel API avec retry automatique
"""
for attempt in range(max_retries):
try:
result = func()
if 'error' not in result:
return result
except Exception as e:
if attempt == max_retries - 1:
return {'error': str(e), 'attempts': max_retries}
# Attente exponentielle
time.sleep(delay * (2 ** attempt))
return {'error': 'Max retries exceeded', 'attempts': max_retries}
# Wrapper pour les appels API
def safe_api_call(url, headers=None, max_retries=3):
"""
Appel API sécurisé avec retry
"""
def make_call():
return fetch_data_from_api(url, headers)
return api_call_with_retry(make_call, max_retries)2. Cache d'API
import time
# Cache simple en mémoire
_api_cache = {}
_cache_duration = 300 # 5 minutes par défaut
def cached_api_call(url, headers=None, cache_duration=None):
"""
Effectue un appel API avec mise en cache
"""
if cache_duration is None:
cache_duration = _cache_duration
cache_key = f"{url}_{str(headers)}"
current_time = time.time()
# Vérifier le cache
if cache_key in _api_cache:
cached_data, timestamp = _api_cache[cache_key]
if current_time - timestamp < cache_duration:
return {**cached_data, '_cached': True}
# Effectuer l'appel
result = fetch_data_from_api(url, headers)
# Mettre en cache si succès
if 'error' not in result:
_api_cache[cache_key] = (result, current_time)
return result
def clear_api_cache():
"""
Vide le cache API
"""
_api_cache.clear()
return {'status': 'Cache cleared'}Sécurité et Bonnes Pratiques
1. Gestion Sécurisée des Clés API
# Ne JAMAIS hardcoder les clés API dans les formules
# Utilisez plutôt des colonnes de configuration
def get_api_key_from_config(service_name, config_table):
"""
Récupère une clé API depuis une table de configuration
"""
for row in config_table:
if row.service == service_name:
return row.api_key
return None
# Dans Grist, créez une table 'API_Config' avec les colonnes:
# - service (texte)
# - api_key (texte, masqué si possible)
# - is_active (booléen)
# Utilisation
# api_key = get_api_key_from_config('openweather', $API_Config_Table)2. Validation des Données
def validate_api_response(response, required_fields):
"""
Valide qu'une réponse API contient tous les champs requis
"""
if 'error' in response:
return {'valid': False, 'error': response['error']}
missing_fields = [field for field in required_fields if field not in response]
if missing_fields:
return {
'valid': False,
'error': f'Champs manquants: {", ".join(missing_fields)}'
}
return {'valid': True, 'data': response}Exemple Complet : Synchronisation de Données
def sync_external_data_to_grist(api_url, mapping_config, last_sync=None):
"""
Synchronise des données externes vers Grist
"""
# Récupérer les données
headers = mapping_config.get('headers', {})
data = fetch_data_from_api(api_url, headers)
if 'error' in data:
return {'success': False, 'error': data['error']}
# Transformer les données selon le mapping
records = data.get(mapping_config.get('data_path', 'data'), [])
transformed_records = []
for record in records:
transformed = {}
for target_field, source_field in mapping_config['field_mapping'].items():
if isinstance(source_field, str):
transformed[target_field] = record.get(source_field)
elif callable(source_field):
transformed[target_field] = source_field(record)
# Filtrer par date si last_sync est fourni
if last_sync and 'updated_at' in record:
if record['updated_at'] <= last_sync:
continue
transformed_records.append(transformed)
return {
'success': True,
'records_synced': len(transformed_records),
'records': transformed_records,
'sync_timestamp': datetime.now().isoformat()
}
# Configuration exemple
sync_config = {
'headers': {'Authorization': 'Bearer YOUR_TOKEN'},
'data_path': 'results',
'field_mapping': {
'customer_name': 'name',
'customer_email': 'email',
'order_count': lambda r: len(r.get('orders', [])),
'total_spent': lambda r: sum(o['amount'] for o in r.get('orders', []))
}
}Conclusion
L'intégration d'API externes transforme Grist en centre de commande de vos données. Que ce soit pour enrichir vos contacts, automatiser des notifications ou synchroniser des systèmes, les possibilités sont infinies.
Pour aller plus loin, explorez notre Guide Complet Python dans Grist et nos Visualisations Avancées.