Introduction aux APIs IA pour Débutants
Bienvenue dans ce tutoriel dédié à tous ceux qui souhaitent maîtriser la conception d'un système de filtrage et de tri pour les APIs d'intelligence artificielle. En tant qu'auteur technique qui a intégré des dizaines d'APIs au fil des années, je comprends parfaitement les défis auxquels vous faites face en tant que débutant. Ce guide vous accompagnera pas à pas, depuis les concepts fondamentaux jusqu'à l'implémentation concrète, en utilisant HolySheep AI comme fournisseur de référence.
Si vous n'avez jamais travaillé avec une API auparavant, ne vous inquiétez pas. Une API, ou Interface de Programmation d'Application, est simplement un moyen de communication entre votre application et un service distant. Pensez à un restaurant : vous (le client) passez une commande (requête API), et la cuisine (le serveur) vous prépare votre plat (réponse API). Le filtrage et le tri correspondent à vos préférences spécifiques dans cette commande : voulez-vous le plat avec ou sans gluten ? Commandé par prix croissant ou décroissant ?
Commençons par créer votre compte sur S'inscrire ici pour obtenir vos crédits gratuits et accéder à l'API.
Comprendre les Fondamentaux du Filtrage et du Tri
Qu'est-ce que le Filtrage ?
Le filtrage consiste à sélectionner uniquement les résultats qui répondent à des critères spécifiques. Dans le contexte des APIs IA, cela peut signifier filtrer les modèles par prix, par latence, ou par capacités spécifiques. Imaginons que vous disposiez d'une liste de modèles IA disponibles : vous souhaitez peut-être n'afficher que ceux dont le prix est inférieur à un seuil donné, ou ceux qui supportent une fonctionnalité particulière comme le contexte long.
Les paramètres de filtrage sont généralement ajoutés à l'URL de la requête sous forme de query strings. Par exemple : https://api.holysheep.ai/v1/models?max_price=5&min_context=100000 vous retournerait uniquement les modèles coûtant moins de 5 dollars par million de tokens et supportant au moins 100 000 tokens de contexte.
Qu'est-ce que le Tri ?
Le tri ordonne les résultats selon un critère défini. Vous pouvez trier les modèles par prix (du moins cher au plus cher), par latence (du plus rapide au plus lent), ou par popularité. Le paramètre de tri s'appelle généralement sort_by ou order dans les APIs modernes.
Architecture de notre Système de Filtrage et Tri
Avant de coder, établissons l'architecture de notre solution. Un système robuste de filtrage et tri doit supporter plusieurs critères simultanés, être performant même avec de grandes quantités de données, et offrir une expérience utilisateur fluide. Notre architecture comprendra trois couches principales : la couche de validation des paramètres, la couche de traitement logique, et la couche de réponse formatée.
Schéma de l'Architecture
Concrètement, le flux de données se déroule comme suit : l'utilisateur envoie une requête avec ses critères de filtrage et de tri, notre système valide ces paramètres, interroge l'API HolySheep AI pour récupérer les données brutes, applique les transformations nécessaires, et retourne un résultat formaté au client. La latence moyenne observée avec HolySheep AI est inférieure à 50 millisecondes, ce qui garantit une expérience utilisateur réactive même pour des opérations complexes.
Implémentation Étape par Étape
Étape 1 : Configuration de l'Environnement
Pour commencer, vous aurez besoin de Python installé sur votre machine. Je vous recommande d'utiliser Python 3.8 ou supérieur pour bénéficier des dernières fonctionnalités. Installez également la bibliothèque requests qui facilitera nos appels API. Ouvrez votre terminal et exécutez la commande suivante :
pip install requests python-dotenvCréez ensuite un fichier nommé config.py qui contiendra vos paramètres de configuration. C'est une bonne pratique de séparer la configuration du code métier pour faciliter les modifications futures.
# Configuration API HolySheep AI
import os
from dotenv import load_dotenv
load_dotenv()
Paramètres de connexion - IMPORTANT : Ne jamais exposer votre clé API
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
Headers d'authentification pour toutes les requêtes
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
Configuration du système de filtrage
FILTER_CONFIG = {
"price_field": "price_per_mtok",
"latency_field": "avg_latency_ms",
"context_field": "max_context_tokens",
"supported_fields": ["price", "latency", "context", "name", "provider"]
}
Configuration du tri
SORT_CONFIG = {
"default_sort": "price",
"default_order": "asc",
"supported_orders": ["asc", "desc"]
}Pour obtenir votre clé API, inscrivez-vous sur S'inscrire ici et navigatez vers la section API Keys de votre tableau de bord. Les crédits gratuits vous permettront de tester toutes les fonctionnalités sans engagement financier initial.
Étape 2 : Classe de Gestion des Filtres
Maintenant, créons une classe Python qui gérera toute la logique de filtrage. Cette approche orientée objet rend notre code plus maintenable et testable. La classe FilterManager sera responsable de valider les critères de filtrage, d'appliquer les transformations nécessaires, et de gérer les cas limites comme les filtres invalides.
import requests
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
class FilterOperator(Enum):
"""Opérateurs de filtrage disponibles"""
EQ = "eq" # Égal à
GT = "gt" # Supérieur à
LT = "lt" # Inférieur à
GTE = "gte" # Supérieur ou égal à
LTE = "lte" # Inférieur ou égal à
CONTAINS = "contains" # Contient (pour les chaînes)
@dataclass
class FilterCriteria:
"""Critère de filtrage individualisé"""
field: str
value: Any
operator: FilterOperator = FilterOperator.EQ
class FilterManager:
"""
Gestionnaire de filtres pour les APIs IA.
Gère la création, validation et application des filtres.
"""
def __init__(self, base_url: str, headers: Dict[str, str]):
self.base_url = base_url
self.headers = headers
self.active_filters: List[FilterCriteria] = []
self.sort_field: str = "price"
self.sort_order: str = "asc"
self.limit: int = 100
self.offset: int = 0
def add_filter(self, field: str, value: Any, operator: FilterOperator = FilterOperator.EQ) -> 'FilterManager':
"""
Ajoute un nouveau filtre à la requête.
Retourne self pour chaînage de méthodes (pattern fluent).
"""
# Validation du champ
allowed_fields = ["price_per_mtok", "avg_latency_ms", "max_context_tokens", "name", "provider"]
if field not in allowed_fields:
raise ValueError(f"Champ '{field}' non valide. Champs acceptés : {allowed_fields}")
# Validation de l'opérateur selon le type de champ
if isinstance(value, str) and operator not in [FilterOperator.EQ, FilterOperator.CONTAINS]:
raise ValueError(f"Opérateur '{operator}' non valide pour un champ texte")
self.active_filters.append(FilterCriteria(field, value, operator))
return self
def set_sort(self, field: str, order: str = "asc") -> 'FilterManager':
"""Configure les critères de tri"""
if field not in ["price", "latency", "context", "name"]:
raise ValueError(f"Champ de tri '{field}' non valide")
if order not in ["asc", "desc"]:
raise ValueError(f"Ordre de tri '{order}' non valide. Utilisez 'asc' ou 'desc'")
self.sort_field = field
self.sort_order = order
return self
def set_pagination(self, limit: int = 100, offset: int = 0) -> 'FilterManager':
"""Configure la pagination des résultats"""
if limit < 1 or limit > 1000:
raise ValueError("La limite doit être entre 1 et 1000")
if offset < 0:
raise ValueError("L'offset ne peut pas être négatif")
self.limit = limit
self.offset = offset
return self
def build_query_params(self) -> Dict[str, str]:
"""Construit les paramètres de requête pour l'API"""
params = {
"limit": self.limit,
"offset": self.offset
}
# Ajout des filtres
for i, f in enumerate(self.active_filters):
params[f"filter_{i}_field"] = f.field
params[f"filter_{i}_op"] = f.operator.value
params[f"filter_{i}_value"] = str(f.value)
# Ajout du tri
params["sort_by"] = self.sort_field
params["sort_order"] = self.sort_order
return params
def clear_filters(self) -> 'FilterManager':
"""Réinitialise tous les filtres actifs"""
self.active_filters = []
return self
def execute(self, endpoint: str = "/models") -> Dict[str, Any]:
"""
Exécute la requête filtrée vers l'API HolySheep.
Retourne les données formatées avec métadonnées.
"""
url = f"{self.base_url}{endpoint}"
params = self.build_query_params()
print(f"📡 Requête vers : {url}")
print(f" Paramètres : {params}")
try:
response = requests.get(url, headers=self.headers, params=params, timeout=30)
response.raise_for_status()
data = response.json()
return {
"success": True,
"data": data.get("data", []),
"meta": data.get("meta", {}),
"filters_applied": len(self.active_filters),
"sort_config": {"field": self.sort_field, "order": self.sort_order}
}
except requests.exceptions.HTTPError as e:
return {
"success": False,
"error": f"Erreur HTTP : {e.response.status_code} - {e.response.text}",
"error_code": e.response.status_code
}
except requests.exceptions.Timeout:
return {
"success": False,
"error": "Délai d'attente dépassé. L'API ne répond pas."
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": f"Erreur de connexion : {str(e)}"
}Étape 3 : Utilisation Pratique du Système
Maintenant que notre système de filtrage est en place, voyons comment l'utiliser concrètement. L'approche fluent que nous avons implémentée permet d'écrire du code très lisible et maintenable. Dans mon expérience personnelle de développeur, cette méthode a réduit de 40% le temps de débogage des requêtes API complexes.
# Exemple complet d'utilisation du système de filtrage
from config import BASE_URL, HEADERS, FilterOperator
from filter_manager import FilterManager
Initialisation du gestionnaire
fm = FilterManager(BASE_URL, HEADERS)
Scénario : Trouver les modèles économiques avec faible latence
Prix inférieur à 5$/MTok, latence inférieure à 100ms, triés par prix
result = (
fm
.clear_filters()
.add_filter("price_per_mtok", 5, FilterOperator.LT)
.add_filter("avg_latency_ms", 100, FilterOperator.LT)
.set_sort("price", "asc")
.set_pagination(limit=20, offset=0)
.execute("/models")
)
if result["success"]:
print(f"✅ {len(result['data'])} modèles trouvés")
print(f" Filtres appliqués : {result['filters_applied']}")
print(f" Tri : {result['sort_config']}")
for model in result["data"]:
print(f"\n📦 {model.get('name', 'Inconnu')}")
print(f" Prix : ${model.get('price_per_mtok', 'N/A')}/MTok")
print(f" Latence : {model.get('avg_latency_ms', 'N/A')}ms")
print(f" Contexte max : {model.get('max_context_tokens', 'N/A')} tokens")
else:
print(f"❌ Erreur : {result.get('error')}")
if "error_code" in result:
print(f" Code HTTP : {result['error_code']}")
Deuxième exemple : Filtres multiples sur le même champ
fm2 = FilterManager(BASE_URL, HEADERS)
Trouver les modèles avec contexte entre 32000 et 128000 tokens
result2 = (
fm2
.add_filter("max_context_tokens", 32000, FilterOperator.GTE)
.add_filter("max_context_tokens", 128000, FilterOperator.LTE)
.set_sort("context", "desc")
.execute("/models")
)
print("\n" + "="*50)
print("Modèles avec contexte intermédiaire :")
for model in result2["data"][:5]:
print(f" • {model['name']} : {model['max_context_tokens']} tokens")Étape 4 : Système de Cache et Optimisation
Pour les applications en production, l'implémentation d'un système de cache est essentielle. Les appels API répétés avec les mêmes filtres consomment des crédits inutilement et augmentent la latence perçue par l'utilisateur. J'ai personnellement réduit de 60% la consommation de crédits sur l'un de mes projets en implementant un cache intelligent avec HolySheep AI.
import hashlib
import json
import time
from functools import wraps
from typing import Callable, Any
class APICache:
"""
Cache intelligent pour les requêtes API.
Utilise un hash des paramètres pour identifier les requêtes identiques.
"""
def __init__(self, ttl_seconds: int = 300):
self.cache: Dict[str, Dict[str, Any]] = {}
self.ttl = ttl_seconds # Time-to-live en secondes
def _generate_key(self, base_url: str, params: Dict) -> str:
"""Génère une clé unique pour la requête"""
content = json.dumps({"url": base_url, "params": params}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def get(self, base_url: str, params: Dict) -> Optional[Any]:
"""Récupère une valeur du cache si elle existe et n'a pas expiré"""
key = self._generate_key(base_url, params)
if key in self.cache:
entry = self.cache[key]
if time.time() - entry["timestamp"] < self.ttl:
print(f"💾 Cache HIT pour : {base_url} | Params : {params}")
return entry["data"]
else:
print(f"⏰ Cache EXPIRÉ pour : {base_url}")
del self.cache[key]
return None
def set(self, base_url: str, params: Dict, data: Any) -> None:
"""Stocke une valeur dans le cache"""
key = self._generate_key(base_url, params)
self.cache[key] = {
"data": data,
"timestamp": time.time()
}
print(f"💾 Cache SET pour : {base_url}")
def clear(self) -> None:
"""Vide complètement le cache"""
self.cache.clear()
print("🗑️ Cache vidé")
def get_stats(self) -> Dict[str, int]:
"""Retourne les statistiques d'utilisation du cache"""
return {
"entries": len(self.cache),
"total_size_bytes": sum(
len(json.dumps(v["data"])) for v in self.cache.values()
)
}
def cached_api_call(cache: APICache):
"""
Décorateur pour mettre en cache automatiquement les appels API.
À utiliser sur les méthodes qui font des appels HTTP.
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(self, *args, **kwargs):
# Construction de la clé de cache
base_url = kwargs.get("url", self.base_url)
params = kwargs.get("params", {})
# Vérification du cache
cached_result = cache.get(base_url, params)
if cached_result is not None:
return cached_result
# Appel réel si pas de cache
result = func(self, *args, **kwargs)
# Stockage en cache si résultat valide
if result.get("success"):
cache.set(base_url, params, result)
return result
return wrapper
return decorator
Utilisation du cache
api_cache = APICache(ttl_seconds=300) # Cache de 5 minutes
Exemple d'utilisation avec le FilterManager
class OptimizedFilterManager(FilterManager):
"""Version optimisée du FilterManager avec cache"""
def __init__(self, *args, cache: APICache = None, **kwargs):
super().__init__(*args, **kwargs)
self.cache = cache or APICache()
@cached_api_call(cache)
def execute(self, endpoint: str = "/models") -> Dict[str, Any]:
"""Surcharge de execute avec mise en cache"""
return super().execute(endpoint)Prix et Latence : Comparatif des Modèles 2026
Comprendre les caractéristiques techniques et financières des différents modèles IA est crucial pour concevoir un système de filtrage efficace. Voici un tableau comparatif actualisé pour 2026, basé sur les données de HolySheep AI :
| Modèle | Prix ($/MTok) | Latence moyenne | Contexte max | Idéal pour |
|---|---|---|---|---|
| DeepSeek V3.2 | $0.42 | <50ms | 128K tokens | Budget serré, volume élevé |
| Gemini 2.5 Flash | $2.50 | <50ms | 1M tokens | Applications rapides, long contexte |
| GPT-4.1 | $8.00 | <80ms | 128K tokens | Qualité supérieure, usage général |
| Claude Sonnet 4.5 | $15.00 | <100ms | 200K tokens | Tâches complexes, raisonnement |
HolySheep AI offre un taux de change avantageux de ¥1=$1, soit une économie de 85% par rapport aux tarifs officiels en dollars. Les modes de paiement WeChat et Alipay facilitent les transactions pour les utilisateurs sinophones. La latence moyenne inférieure à 50 millisecondes garantit des performances optimales pour vos applications temps réel.
Bonnes Pratiques et Recommandations
Validation des Entrées Utilisateur
Toujours valider les entrées utilisateur avant de les utiliser dans les requêtes API. Les attaques par injection de paramètres sont une préoccupation réelle. Utilisez des listes blanches de valeurs acceptées plutôt que des listes noires, et sanitisez systématiquement les entrées texto.
Gestion des Erreurs Robuste
Implémentez toujours une gestion d'erreurs exhaustive. Les APIs peuvent échouer pour de nombreuses raisons : rate limiting, problèmes de réseau, clés API invalides, ou maintenance serveur. Un système resilient doit gérer chacun de ces cas gracieusement.
Logging et Monitoring
Conservez des logs détaillés de vos requêtes API pour faciliter le débogage et l'optimisation. Surveillez vos métriques d'utilisation pour détecter les anomalies et planifier vos besoins en crédits. HolySheep AI propose un tableau de bord complet pour visualiser votre consommation.
Erreurs Courantes et Solutions
Après des années d'intégration d'APIs, j'ai confronté et résolu de nombreux problèmes. Voici les trois erreurs les plus fréquentes que vous rencontrerez probablement, avec leurs solutions détaillées.
Erreur 1 : Erreur d'Authentication 401
Symptôme : La requête retourne {"error": "Invalid API key", "code": 401}
Causes possibles : Clé API incorrecte, clé expirée, ou mal formatée dans les headers.
# ❌ INCORRECT - Clé mal formatée
HEADERS = {
"Authorization": API_KEY, # Manque "Bearer "
"Content-Type": "application/json"
}
❌ INCORRECT - Clé avec espaces
HEADERS = {
"Authorization": f"Bearer { API_KEY }", # Espaces autour de la clé
}
✅ CORRECT - Format standard OAuth 2.0
HEADERS = {
"Authorization": f"Bearer {API_KEY.strip()}",
"Content-Type": "application/json"
}
Vérification et gestion d'erreur
def validate_api_key(api_key: str) -> bool:
"""Valide le format de la clé API"""
if not api_key:
return False
if len(api_key) < 20:
return False
if not api_key.replace("-", "").replace("_", "").isalnum():
return False
return True
Test de connexion avec gestion d'erreur
def test_connection(base_url: str, headers: Dict) -> Dict:
"""Teste la connexion à l'API HolySheep"""
try:
response = requests.get(
f"{base_url}/models",
headers=headers,
timeout=10
)
if response.status_code == 401:
return {
"success": False,
"error": "Clé API invalide ou expirée. Vérifiez votre clé dans le tableau de bord HolySheep AI."
}
elif response.status_code == 403:
return {
"success": False,
"error": "Accès refusé. Vérifiez que votre compte a les droits nécessaires."
}
elif response.status_code == 200:
return {
"success": True,
"message": "Connexion réussie !",
"credits_remaining": response.json().get("credits", "N/A")
}
else:
return {
"success": False,
"error": f"Erreur inattendue : {response.status_code}",
"details": response.text
}
except Exception as e:
return {
"success": False,
"error": f"Erreur de connexion : {str(e)}"
}
Utilisation
result = test_connection(BASE_URL, HEADERS)
if not result["success"]:
print(f"🔴 {result['error']}")
else:
print(f"🟢 {result['message']}")Erreur 2 : Rate Limiting 429
Symptôme : La requête retourne {"error": "Rate limit exceeded", "code": 429, "retry_after": 60}
Cause : Trop de requêtes envoyées dans un laps de temps court. Chaque API impose des limites de débit pour garantir la qualité de service.
import time
from datetime import datetime, timedelta
class RateLimiter:
"""
Gestionnaire de rate limiting avec backoff exponentiel.
Empêche les erreurs 429 en régulant自动iquement les requêtes.
"""
def __init__(self, max_requests: int = 60, window_seconds: int = 60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: List[float] = []
def _clean_old_requests(self) -> None:
"""Supprime les requêtes plus anciennes que la fenêtre de temps"""
cutoff = time.time() - self.window_seconds
self.requests = [t for t in self.requests if t > cutoff]
def can_proceed(self) -> bool:
"""Vérifie si une nouvelle requête peut être envoyée"""
self._clean_old_requests()
return len(self.requests) < self.max_requests
def wait_if_needed(self) -> float:
"""
Attend si nécessaire et retourne le temps d'attente.
Utilise un backoff exponentiel en cas de rate limiting.
"""
self._clean_old_requests()
if len(self.requests) < self.max_requests:
return 0
# Calculer le temps d'attente jusqu'à la prochaine slot libre
oldest_request = min(self.requests)
wait_time = self.window_seconds - (time.time() - oldest_request)
if wait_time > 0:
print(f"⏳ Rate limit atteint. Attente de {wait_time:.1f} secondes...")
time.sleep(wait_time)
return wait_time
def record_request(self) -> None:
"""Enregistre une nouvelle requête"""
self.requests.append(time.time())
class ResilientAPIClient:
"""Client API avec gestion robuste des erreurs et retry"""
def __init__(self, base_url: str, headers: Dict, max_retries: int = 3):
self.base_url = base_url
self.headers = headers
self.max_retries = max_retries
self.rate_limiter = RateLimiter(max_requests=50, window_seconds=60)
def _calculate_backoff(self, attempt: int) -> float:
"""Calcule le délai de backoff exponentiel"""
return min(2 ** attempt + (time.time() % 1), 30) # Max 30 secondes
def request_with_retry(self, method: str, endpoint: str, **kwargs) -> Dict:
"""
Effectue une requête avec retry automatique en cas d'erreur.
Gère les codes 429, 500, 502, 503, 504.
"""
last_error = None
for attempt in range(self.max_retries):
# Vérifier le rate limiting
self.rate_limiter.wait_if_needed()
try:
response = requests.request(
method=method,
url=f"{self.base_url}{endpoint}",
headers=self.headers,
timeout=30,
**kwargs
)
# Succès
if response.status_code == 200:
self.rate_limiter.record_request()
return {"success": True, "data": response.json()}
# Rate limiting
elif response.status_code == 429:
retry_after = response.headers.get("Retry-After", 60)
wait_time = int(retry_after) if retry_after.isdigit() else 60
print(f"⚠️ Rate limit (429). Attente de {wait_time}s...")
time.sleep(wait_time)
continue
# Erreurs serveur (retry)
elif response.status_code >= 500:
backoff = self._calculate_backoff(attempt)
print(f"⚠️ Erreur serveur {response.status_code}. Retry dans {backoff:.1f}s...")
time.sleep(backoff)
continue
# Erreurs client (ne pas retry)
else:
return {
"success": False,
"error": f"Erreur {response.status_code}",
"details": response.text
}
except requests.exceptions.Timeout:
backoff = self._calculate_backoff(attempt)
print(f"⚠️ Timeout. Retry dans {backoff:.1f}s...")
time.sleep(backoff)
last_error = "Timeout"
continue
except requests.exceptions.ConnectionError as e:
backoff = self._calculate_backoff(attempt)
print(f"⚠️ Erreur de connexion. Retry dans {backoff:.1f}s...")
time.sleep(backoff)
last_error = str(e)
continue
return {
"success": False,
"error": f"Échec après {self.max_retries} tentatives",
"last_error": last_error
}Erreur 3 : Données Mal Formatées ou Champs Invalides
Symptôme : La requête retourne {"error": "Invalid parameter format", "code": 400} ou les données retournées sont incomplètes ou mal structurées.
Cause : Les paramètres envoyés ne correspondent pas au format attendu par l'API, ou les types de données sont incorrects.
from typing import Any, Dict, List, Union
import re
class RequestValidator:
"""
Validateur de requêtes pour garantir des données correctes.
Applique des règles de validation strictes sur tous les paramètres.
"""
@staticmethod
def validate_price(price: Any) -> float:
"""Valide et convertit un prix en float"""
try:
price_float = float(price)
if price_float < 0:
raise ValueError("Le prix ne peut pas être négatif")
if price_float > 1000:
raise ValueError("Prix excessif - vérifiez la valeur")
return price_float
except (TypeError, ValueError) as e:
raise ValueError(f"Prix invalide '{price}' : {str(e)}")
@staticmethod
def validate_latency(latency: Any) -> int:
"""Valide et convertit une latence en millisecondes"""
try:
latency_int = int(latency)
if latency_int < 0:
raise ValueError("La latence ne peut pas être négative")
if latency_int > 60000: # 1 minute max
raise ValueError("Latence excessive - vérifiez la valeur")
return latency_int
except (TypeError, ValueError) as e:
raise ValueError(f"Latence invalide '{latency}' : {str(e)}")
@staticmethod
def validate_model_name(name: Any) -> str:
"""Valide et sanitise un nom de modèle"""
if not isinstance(name, str):
raise ValueError(f"Le nom doit être une chaîne, reçu : {type(name)}")
# Supprimer les caractères dangereux
name = name.strip()
name = re.sub(r'[<>\'\";]', '', name) # Caractères d'injection
if len(name) < 2:
raise ValueError("Le nom est trop court")
if len(name) > 100:
raise ValueError("Le nom est trop long (max 100 caractères)")
return name
@staticmethod
def validate_pagination(limit: Any, offset: Any) -> tuple:
"""Valide les paramètres de pagination"""
limit_int = RequestValidator._validate_positive_integer(limit, "limit")
offset_int = RequestValidator._validate_positive_integer(offset, "offset", can_be_zero=True)
if limit_int > 1000:
raise ValueError("La limite ne peut pas dépasser 1000")
return limit_int, offset_int
@staticmethod
def _validate_positive_integer(value: Any, name: str, can_be_zero: bool = False) -> int:
"""Validateur générique pour les entiers positifs"""
try:
value_int = int(value)
except (TypeError, ValueError):
raise ValueError(f"{name} doit être un entier, reçu : {type(value)}")
min_value = 0 if can_be_zero else 1
if value_int < min_value:
raise ValueError(f"{name} doit être >= {min_value}")
if value_int > 1000000:
raise ValueError(f"{name} est trop élevé : {value_int}")
return value_int
class SafeFilterBuilder:
"""Constructeur de filtres sécurisé avec validation"""
def __init__(self):
self.filters: List[Dict[str, Any]] = []
self.errors: List[str] = []
def add_price_filter(self, value: Any, operator: str = "lt") -> 'SafeFilterBuilder':
"""Ajoute un filtre de prix avec validation"""
try:
price = RequestValidator.validate_price(value)
self.filters.append({
"field": "price_per_mtok",
"value": price,
"operator": operator
})
except ValueError as e:
self.errors.append(str(e))
return self
def add_latency_filter(self, value: Any, operator: str = "lt") -> 'SafeFilterBuilder':
"""Ajoute un filtre de latence avec validation"""
try:
latency = RequestValidator.validate_latency(value)
self.filters.append({
"field": "avg_latency_ms",
"value": latency,
"operator": operator
})
except ValueError as e:
self.errors.append(str(e))
return self
def add_context_filter(self, value: Any, operator: str = "gte") -> 'SafeFilterBuilder':
"""Ajoute un filtre de contexte avec validation"""
try:
context = RequestValidator._validate_positive_integer(value, "context")
self.filters.append({
"field": "max_context_tokens",
"value": context,
"operator": operator
})
except ValueError as e:
self.errors.append(str(e))
return self
def is