Les décorateurs Python permettent d ajouter des comportements transversaux aux fonctions de pipeline sans les polluer. En entretien, on évalue la capacité à écrire des décorateurs robustes et utiles.
Qu est-ce qu un décorateur Python ? Comment fonctionne-t-il ?
import functools
# Un décorateur est une fonction qui prend une fonction et retourne une fonction
def mon_decorateur(func):
@functools.wraps(func) # préserver le nom et la docstring
def wrapper(*args, **kwargs):
print(f'Avant {func.__name__}')
result = func(*args, **kwargs)
print(f'Apres {func.__name__}')
return result
return wrapper
@mon_decorateur
def extract_data(source: str) -> list:
return []
# Equivalent à : extract_data = mon_decorateur(extract_data)Comment écrivez-vous un décorateur retry avec backoff exponentiel ?
import functools
import time
import logging
def retry(max_attempts=3, backoff_factor=2, exceptions=(Exception,)):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
wait = backoff_factor ** attempt
logging.warning(
f'{func.__name__} failed (attempt {attempt+1}/{max_attempts}). '
f'Retrying in {wait}s. Error: {e}'
)
time.sleep(wait)
raise last_exception
return wrapper
return decorator
@retry(max_attempts=3, backoff_factor=2, exceptions=(requests.Timeout, requests.ConnectionError))
def fetch_api_data(url: str) -> dict:
return requests.get(url, timeout=30).json()Comment cachéz-vous le résultat d une requête SQL coûteuse avec un décorateur ?
import functools
import time
from typing import Optional
def cache_with_ttl(ttl_seconds: int = 3600):
def decorator(func):
cache = {}
@functools.wraps(func)
def wrapper(*args, **kwargs):
# Clé de cache basée sur les arguments
key = str(args) + str(sorted(kwargs.items()))
now = time.time()
if key in cache:
result, timestamp = cache[key]
if now - timestamp < ttl_seconds:
return result
result = func(*args, **kwargs)
cache[key] = (result, now)
return result
wrapper.cache_clear = lambda: cache.clear()
return wrapper
return decorator
@cache_with_ttl(ttl_seconds=3600)
def get_reference_table(table_name: str) -> pd.DataFrame:
return run_query(f'SELECT * FROM {table_name}')Comment loguez-vous automatiquement les entrées, sorties et durées d exécution ?
import functools
import time
import logging
def pipeline_step(log_inputs=True, log_output_size=True):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
logger = logging.getLogger(func.__module__)
if log_inputs:
logger.info(f'START {func.__name__} | args={args[:2]} kwargs={list(kwargs.keys())}')
try:
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
size_info = ''
if log_output_size and hasattr(result, '__len__'):
size_info = f' | output_size={len(result)}'
logger.info(f'END {func.__name__} | elapsed={elapsed:.2f}s{size_info}')
return result
except Exception as e:
logger.error(f'FAILED {func.__name__} | elapsed={time.perf_counter()-start:.2f}s | error={e}')
raise
return wrapper
return decoratorComment validez-vous les paramètres d une fonction de pipeline sans polluer son code ?
import functools
from datetime import date
def validate_date_range(start_param='start_date', end_param='end_date', max_days=365):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = kwargs.get(start_param)
end = kwargs.get(end_param)
if start and end:
if start > end:
raise ValueError(f'{start_param} must be <= {end_param}')
delta = (end - start).days
if delta > max_days:
raise ValueError(f'Date range {delta} days exceeds max {max_days}')
return func(*args, **kwargs)
return wrapper
return decorator
@validate_date_range(max_days=90)
def extract_orders(start_date: date, end_date: date) -> list:
return query_orders(start_date, end_date)Dans quel ordre s appliquent plusieurs décorateurs ? Comment les composer correctement ?
@retry(max_attempts=3)
@cache_with_ttl(ttl_seconds=300)
@pipeline_step()
def fetch_and_cache(url: str) -> dict:
return requests.get(url).json()
# Ordre d application : de bas en haut
# fetch_and_cache = retry(cache_with_ttl(pipeline_step(func)))
# Appel : retry -> cache_with_ttl -> pipeline_step -> func
# Règle : mettre retry en extérieur (première couche)
# pour que les retries bénéficient du cache
# Si cache en extérieur, les retries ne sont pas déclenchés
# car le cache retourne directement| Niveau | Maitrise | Signal GO | NO-GO |
|---|---|---|---|
| Confirmé | Écrire un décorateur simple avec wraps, comprend l ordre | A écrit un décorateur retry ou cache, utilise functools.wraps | Ne sait pas ce qu est un décorateur |
| Senior | Décorateurs paramétrés, composition, décorateurs de classe | Écrit des décorateurs paramétrés, gère l ordre de composition | N utilise jamais functools.wraps |
Premier entretien gratuit. Rapport GO/NO-GO sous 48h.