---
name: add-dlt-rest-api-source
description: Implement a REST API source using dlt's declarative REST API framework. Use when the user wants to add a REST API source, configure rest_api_source, set up pagination (cursor, offset, page number, header link), configure incremental loading, or define declarative API config with authentication.
---

# Add a REST API Source

Implement a REST API source for a dlt connector using dlt's declarative REST API framework.

## Context

dlt provides a declarative REST API framework that handles pagination, authentication, and incremental loading through configuration rather than custom code. Choose the simplest pattern that meets your needs.

## Steps

1. Determine the appropriate pattern based on API complexity.

   | Complexity | Pattern | When to Use |
   |------------|---------|-------------|
   | Simple | `rest_api_source` | Public APIs, no auth, simple pagination |
   | Medium | `@dlt.source` + `rest_api_resources` | Auth required, custom config |
   | Complex | Custom `@dlt.source` module | OAuth, complex transformations, multiple API versions |

2. For simple public APIs, use the declarative `rest_api_source` pattern.

   ```python
   from dlt.sources.rest_api import rest_api_source

   source = rest_api_source({
       "client": {
           "base_url": "https://api.example.com/",
       },
       "resource_defaults": {
           "endpoint": {"params": {"limit": 100}},
           "write_disposition": "replace",
       },
       "resources": [
           "users",
           "posts",
           "comments",
       ],
   })
   ```

3. For APIs requiring authentication, use `@dlt.source` with `rest_api_resources`. Name the source function `<connector_name>_source`.

   ```python
   import dlt
   from dlt.sources.rest_api import RESTAPIConfig, rest_api_resources

   @dlt.source
   def <connector_name>_source(api_token: str = dlt.secrets.value):
       config: RESTAPIConfig = {
           "client": {
               "base_url": "https://api.example.com/",
               "auth": {
                   "type": "api_key",
                   "name": "Authorization",
                   "api_key": f"Bearer {api_token}",
                   "location": "header",
               },
           },
           "resource_defaults": {
               "primary_key": "id",
               "write_disposition": "merge",
           },
           "resources": [
               {
                   "name": "users",
                   "endpoint": {
                       "path": "v1/users",
                       "method": "GET",
                       "data_selector": "data",
                   },
               },
           ],
       }
       yield from rest_api_resources(config)
   ```

   For **OAuth2 client credentials** (common with enterprise APIs like HubSpot, Salesforce):

   ```python
   "client": {
       "base_url": "https://api.example.com/",
       "auth": {
           "type": "oauth2_client_credentials",
           "access_token_url": "https://api.example.com/oauth/token",
           "client_id": dlt.secrets["sources.<connector_name>.client_id"],
           "client_secret": dlt.secrets["sources.<connector_name>.client_secret"],
       },
   },
   ```

4. Configure pagination for each resource as needed.

   **Cursor-based pagination:**
   ```python
   "paginator": {
       "type": "cursor",
       "cursor_path": "meta.next_cursor",
       "cursor_param": "cursor",
   }
   ```

   **Offset-based pagination:**
   ```python
   "paginator": {
       "type": "offset",
       "limit": 100,
       "offset": 0,
       "offset_param": "start",
       "limit_param": "limit",
       "total_path": "meta.total",
   }
   ```

   **Page number pagination:**
   ```python
   "paginator": {
       "type": "page_number",
       "page_param": "page",
       "total_path": "meta.total_pages",
   }
   ```

   **Link header pagination:**
   ```python
   "paginator": "header_link"
   ```

   **Single-page (no pagination):**
   ```python
   "paginator": "single_page"
   ```
   Use for endpoints that return a single object or a complete list in one response.

5. Configure incremental loading for resources that support it.

   ```python
   {
       "name": "orders",
       "endpoint": {
           "path": "v1/orders",
           "params": {
               "updated_since": "{incremental.start_value}",
           },
           "data_selector": "data",
           "incremental": {
               "cursor_path": "updated_at",
               "initial_value": "2024-01-01T00:00:00Z",
           },
       },
       "primary_key": "id",
       "write_disposition": "merge",
   }
   ```

6. Add the source credentials to `.dlt/secrets_example.toml`.

   ```toml
   [sources.<source_name>]
   api_token = "your-api-token-here"
   ```

   The `dlt.secrets.value` default in the source function signature tells dlt to look up the value at `[sources.<DLT_SOURCE_NAME>.api_token]` in secrets.toml.

7. Wire the source into the `run_pipeline()` function.

   ```python
   source = <connector_name>_source()
   load_info = pipeline.run(source)
   ```

8. Test the source implementation.

   ```bash
   uv sync
   uv run --env-file=.env python <connector_name>_pipeline.py
   ```

## Validation

- [ ] Source yields data for all configured resources
- [ ] Pagination fetches all pages (not just the first)
- [ ] Incremental loading only fetches new/updated records on subsequent runs
- [ ] Authentication works correctly
- [ ] `data_selector` matches the API response structure

## Examples

**Write disposition decision tree:**

```
Is data immutable (e.g., events, logs)?
├── Yes → write_disposition: "append"
└── No → Can records be updated?
    ├── Yes → write_disposition: "merge" (requires primary_key)
    └── No → write_disposition: "replace"
```

**Common issues:**

| Issue | Likely Cause |
|-------|--------------|
| Empty results | `data_selector` doesn't match API response structure |
| Auth failures | Token format wrong (some APIs need "Bearer " prefix) |
| Pagination stops early | `total_path` or cursor path incorrect |
| Incremental not working | `cursor_path` field doesn't exist in response |