Accueil›Blog›Test technique Python décorateurs pour la data : caching, retry, logging

Guide recrutement data

Test technique Python décorateurs pour la data : caching, retry, logging

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.

Data Builder·Juin 2025·6 min de lecture·Data Engineer

Sommaire

Fonctionnement des décorateurs
Décorateur retry
Décorateur cache
Logging automatique
Validation des inputs
Composition de décorateurs
Grille

1Fonctionnement des décorateurs

Question discriminante

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)

@functools.wraps — obligatoire pour préserver le nom, la docstring et la signature de la fonction décorée
*args, **kwargs — permet au wrapper d accepter n importe quels arguments

2Décorateur retry pour les appels réseau

Question discriminante

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()

3Décorateur cache pour les données coûteuses

Question discriminante

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}')

4Logging automatique des pipelines

Question discriminante

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 decorator

5Validation des inputs avec décorateurs

Question discriminante

Comment 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)

6Composer plusieurs décorateurs

Question discriminante

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

Ordre critique — le décorateur le plus externe s exécute en premier à l appel
Retry en externe — le retry doit envelopper le cache pour que les erreurs déclenchent des retries

import functools, time, logging

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1: raise
                    time.sleep(delay * (2 ** attempt))
        return wrapper
    return decorator

def timed_cache(seconds=300):
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        def wrapper(*args):
            import datetime
            now = datetime.datetime.now()
            if args in cache:
                result, ts = cache[args]
                if (now - ts).seconds < seconds:
                    return result
            result = func(*args)
            cache[args] = (result, now)
            return result
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0, exceptions=(ConnectionError,))
@timed_cache(seconds=600)
def fetch_data(url: str) -> dict:
    return requests.get(url).json()

functools.wraps - preserve le nom et la docstring. Sans wraps, func.__name__ retourne wrapper au lieu du nom reel
Decorateurs en production data - retry pour les appels API instables, caching pour les lookups frequents, logging pour tracer les appels
Chaining de decorateurs - ordre important : @retry @cache signifie retry(cache(func)). Le decorateur le plus proche de def s applique en premier
Decorateurs de classe - @dataclass, @property, @classmethod, @staticmethod sont tous des decorateurs natifs Python. Meme mecanique
Decorateur avec arguments - triple imbrication : decorator_factory(args) retourne decorator(func) qui retourne wrapper(*args). Pattern standard

import functools, time, logging
from typing import Callable, Any

def retry(max_attempts=3, delay=1.0, backoff=2.0, exceptions=(Exception,)):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exc = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts - 1:
                        wait = delay * (backoff ** attempt)
                        logging.warning(f"{func.__name__} failed ({attempt+1}/{max_attempts}): retry in {wait:.1f}s")
                        time.sleep(wait)
            raise last_exc
        return wrapper
    return decorator

def timed_cache(ttl_seconds=300):
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        def wrapper(*args):
            import time as t
            now = t.monotonic()
            if args in cache:
                result, ts = cache[args]
                if now - ts < ttl_seconds:
                    return result
            result = func(*args)
            cache[args] = (result, now)
            return result
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0, exceptions=(ConnectionError, TimeoutError))
@timed_cache(ttl_seconds=600)
def fetch_rate(currency: str) -> float:
    return requests.get(f"https://api.rates.io/{currency}").json()['rate']

functools.wraps obligatoire - preserve le nom et la docstring. Sans wraps, func.__name__ retourne 'wrapper' ce qui casse les logs et le debugging
Stacking de decorateurs - ordre important : @retry @cache signifie retry(cache(f)). Le decorateur le plus proche de def s applique en premier
Decorateurs de classe - @dataclass, @property, @classmethod, @staticmethod sont tous des decorateurs natifs Python, meme mecanique
Cas data concrets - retry pour les appels API instables, timed_cache pour les lookups referentiels, logging/timing pour les pipelines
Context manager vs decorateur - une fonction qui gere une ressource (connexion DB) = preferer context manager. Une fonction qui modifie le comportement = decorateur

7Grille par niveau

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

1How decorators work

Discriminating question

What is a Python decorator? How does it work?

import functools

# A decorator is a function that takes a function and returns a function
def mon_decorateur(func):
    @functools.wraps(func)  # preserve the name and docstring
    def wrapper(*args, **kwargs):
        print(f'Before {func.__name__}')
        result = func(*args, **kwargs)
        print(f'After {func.__name__}')
        return result
    return wrapper

@mon_decorateur
def extract_data(source: str) -> list:
    return []

# Equivalent to: extract_data = mon_decorateur(extract_data)

@functools.wraps — mandatory to preserve the name, docstring and signature of the decorated function
*args, **kwargs — allows the wrapper to accept any arguments

2Retry decorator for network calls

Discriminating question

How do you write a retry decorator with exponential backoff?

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()

3Cache decorator for expensive data

Discriminating question

How do you cache the result of an expensive SQL query with a decorator?

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):
            # Cache key based on 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}')

4Automatic pipeline logging

Discriminating question

How do you automatically log inputs, outputs and execution times?

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 decorator

5Input validation with decorators

Discriminating question

How do you validate pipeline function parameters without polluting its 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)

6Composing multiple decorators

Discriminating question

In what order do multiple decorators apply? How do you compose them correctly?

@retry(max_attempts=3)
@cache_with_ttl(ttl_seconds=300)
@pipeline_step()
def fetch_and_cache(url: str) -> dict:
    return requests.get(url).json()

# Application order: bottom to top
# fetch_and_cache = retry(cache_with_ttl(pipeline_step(func)))
# Call: retry -> cache_with_ttl -> pipeline_step -> func

# Rule: put retry on the outside (first layer)
# so that retries benefit from the cache
# If cache is on the outside, retries are not triggered
# because the cache returns directly

Critical order — the outermost decorator executes first at call time
Retry on the outside — retry must wrap the cache so that errors trigger retries

import functools, time, logging

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1: raise
                    time.sleep(delay * (2 ** attempt))
        return wrapper
    return decorator

def timed_cache(seconds=300):
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        def wrapper(*args):
            import datetime
            now = datetime.datetime.now()
            if args in cache:
                result, ts = cache[args]
                if (now - ts).seconds < seconds:
                    return result
            result = func(*args)
            cache[args] = (result, now)
            return result
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0, exceptions=(ConnectionError,))
@timed_cache(seconds=600)
def fetch_data(url: str) -> dict:
    return requests.get(url).json()

functools.wraps - preserves the name and docstring. Without wraps, func.__name__ returns wrapper instead of the real name
Decorators in production data - retry for unstable API calls, caching for frequent lookups, logging to trace calls
Decorator chaining - order matters: @retry @cache means retry(cache(func)). The decorator closest to def applies first
Class decorators - @dataclass, @property, @classmethod, @staticmethod are all native Python decorators. Same mechanism
Decorator with arguments - triple nesting: decorator_factory(args) returns decorator(func) which returns wrapper(*args). Standard pattern

import functools, time, logging
from typing import Callable, Any

def retry(max_attempts=3, delay=1.0, backoff=2.0, exceptions=(Exception,)):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exc = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts - 1:
                        wait = delay * (backoff ** attempt)
                        logging.warning(f"{func.__name__} failed ({attempt+1}/{max_attempts}): retry in {wait:.1f}s")
                        time.sleep(wait)
            raise last_exc
        return wrapper
    return decorator

def timed_cache(ttl_seconds=300):
    def decorator(func):
        cache = {}
        @functools.wraps(func)
        def wrapper(*args):
            import time as t
            now = t.monotonic()
            if args in cache:
                result, ts = cache[args]
                if now - ts < ttl_seconds:
                    return result
            result = func(*args)
            cache[args] = (result, now)
            return result
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0, exceptions=(ConnectionError, TimeoutError))
@timed_cache(ttl_seconds=600)
def fetch_rate(currency: str) -> float:
    return requests.get(f"https://api.rates.io/{currency}").json()['rate']

functools.wraps mandatory - preserves the name and docstring. Without wraps, func.__name__ returns 'wrapper' which breaks logs and debugging
Decorator stacking - order matters: @retry @cache means retry(cache(f)). The decorator closest to def applies first
Class decorators - @dataclass, @property, @classmethod, @staticmethod are all native Python decorators, same mechanism
Concrete data use cases - retry for unstable API calls, timed_cache for reference lookups, logging/timing for pipelines
Context manager vs decorator - a function that manages a resource (DB connection) = prefer context manager. A function that modifies behavior = decorator

7Level grid

Level	Mastery	GO signal	NO-GO
Confirmed	Write a simple decorator with wraps, understands the order	Has written a retry or cache decorator, uses functools.wraps	Does not know what a decorator is
Senior	Parameterized decorators, composition, class decorators	Writes parameterized decorators, manages composition order	N Vous recrutez un Data Engineer Python ? Premier entretien gratuit. Rapport GO/NO-GO sous 48h. Tester gratuitement Reserver un appel Comment ca marche A propos Blog CGU contact@databuilder.io © 2025 Data Builder

Test technique Python décorateurs pour la data : caching, retry, logging

1Fonctionnement des décorateurs

2Décorateur retry pour les appels réseau

3Décorateur cache pour les données coûteuses

4Logging automatique des pipelines

5Validation des inputs avec décorateurs

6Composer plusieurs décorateurs

7Grille par niveau

1How decorators work

2Retry decorator for network calls

3Cache decorator for expensive data

4Automatic pipeline logging

5Input validation with decorators

6Composing multiple decorators

7Level grid

Vous recrutez un Data Engineer Python ?