Patterns de pagination dlt

dlt fournit un système de paginateurs pour gérer les réponses API multi-pages. Vous sélectionnez un paginateur correspondant au style de pagination de votre API, le configurez et l’attachez à un RESTClient. La méthode paginate() gère l’itération à partir de là.

Paginateurs intégrés

dlt embarque des paginateurs pour les patterns que vous rencontrerez dans les APIs réelles :

JSONLinkPaginator — Pour les APIs qui retournent l’URL de la page suivante dans le corps JSON. Courant dans les APIs REST modernes.

from dlt.sources.helpers.rest_client.paginators import JSONLinkPaginator

# L'API retourne : {"data": [...], "next": "https://api.example.com/v1/items?cursor=abc123"}
client = RESTClient(
    base_url="https://api.example.com/v1",
    paginator=JSONLinkPaginator(next_url_path="next")
)

Le next_url_path est une expression JSONPath pointant vers l’URL de la page suivante dans le corps de la réponse.

HeaderLinkPaginator — Pour les APIs qui retournent l’URL de la page suivante dans le header de réponse Link. L’API de GitHub utilise ce pattern (Link: <https://api.github.com/...>; rel="next"). Aucune configuration nécessaire au-delà de l’instanciation — le paginateur analyse le format de header de lien RFC 5988 standard.

OffsetPaginator — Pagination classique offset/limit. Incrémente le paramètre offset par la taille de page après chaque réponse.

from dlt.sources.helpers.rest_client.paginators import OffsetPaginator

paginator = OffsetPaginator(limit=100)
# Génère : ?limit=100&offset=0, ?limit=100&offset=100, ?limit=100&offset=200...

S’arrête quand l’API retourne moins d’enregistrements que la limite (signalant la dernière page) ou quand la réponse ne contient aucun enregistrement.

PageNumberPaginator — Incréments simples de numéro de page. Pour les APIs avec ?page=1, ?page=2, etc.

JSONResponseCursorPaginator — Pour les APIs qui retournent un token de curseur opaque dans le corps de la réponse. Le paginateur extrait le curseur, le passe en paramètre de requête dans la prochaine requête, et s’arrête quand aucun curseur n’est retourné.

Paginateurs personnalisés

Quand aucun des paginateurs intégrés ne convient, étendez BasePaginator. Vous implémentez deux méthodes :

• update_state() : Appelée après chaque réponse. Analysez la réponse et extrayez l’état de pagination que vous devez conserver.
• update_request() : Appelée avant chaque requête. Modifiez la requête sortante pour inclure l’état de pagination.

from dlt.sources.helpers.rest_client.paginators import BasePaginator

class ProprietaryHeaderPaginator(BasePaginator):
    def __init__(self):
        super().__init__()
        self._next_token = None

    def update_state(self, response):
        # Extraire le token de pagination depuis un header personnalisé
        self._next_token = response.headers.get("X-Next-Page-Token")
        if not self._next_token:
            self._has_next_page = False

    def update_request(self, request):
        if self._next_token:
            request.params["page_token"] = self._next_token

Le framework appelle ces méthodes en séquence, gère la boucle et traite la condition d’arrêt quand _has_next_page est False. Votre code se concentre sur la logique d’analyse spécifique à l’API, pas sur la boucle de pagination elle-même.

Cette approche accommode des patterns non standard tels que les formats propriétaires dans les headers de réponse, l’état de pagination à champs multiples, ou les APIs qui signalent la dernière page via un champ de statut dans le corps plutôt qu’un tableau de résultats vide.

Configurer les paginateurs dans REST API Source

Lors de l’utilisation de REST API Source plutôt que de RESTClient directement, les paginateurs sont configurés de manière déclarative par endpoint :

{
    "name": "orders",
    "endpoint": {
        "path": "orders",
        "paginator": {
            "type": "offset",
            "limit": 100
        }
    }
}

Le champ type correspond aux paginateurs intégrés : "json_link", "header_link", "offset", "page_number", "cursor". Vous pouvez également passer des instances de paginateurs personnalisés pour les endpoints nécessitant un comportement non standard.

Déboguer les problèmes de pagination

Le problème de pagination le plus courant : vous obtenez exactement une page de résultats quand vous en attendez plusieurs. Cela signifie presque toujours que le paginateur n’identifie pas correctement le signal “il y a d’autres pages”.

Étapes de débogage :

1. Testez d’abord avec une limite réduite. Définissez limit=2 ou limit=5 pour vérifier que la pagination s’active même avec un petit dataset.
2. Inspectez une réponse brute. Utilisez requests directement pour récupérer une page et examinez le corps de réponse et les headers complets — trouvez où se trouve réellement le signal de page suivante.
3. Vérifiez la condition d’arrêt du paginateur. OffsetPaginator s’arrête quand len(records) < limit. Si votre API retourne un nombre fixe d’enregistrements même sur la dernière page, vous avez besoin d’un type de paginateur différent.
4. Utilisez des noms de pipeline uniques lors du test des variations. dlt stocke l’état par nom de pipeline — si vous testez le chargement incrémental en parallèle avec la pagination, les conflits de noms peuvent causer des problèmes d’état surprenants.

Quand la pagination fonctionne correctement, la boucle paginate() s’exécute silencieusement jusqu’à la fin. Quand elle ne fonctionne pas, la boucle se termine après une itération et vous n’avez chargé que la première page. Ce mode d’échec est facile à détecter et généralement facile à corriger une fois que vous avez lu la réponse API réelle.