BlogGithub

Dynamic Endpoints ⚡

Examples of using dynamic endpoints to programmatically generate multiple endpoint configurations

Dynamic endpoints allow you to programmatically generate multiple endpoint configurations based on runtime data. This is powerful for APIs where the list of available endpoints or resources isn't known until you query the API itself, or when you need to create many similar endpoints without repeating configuration.

Learn more: Dynamic Endpoints " State Variables " Processors

How Dynamic Endpoints Work

Dynamic endpoints use the dynamic_endpoints key to generate multiple endpoint configurations:

  1. Setup Phase (optional): Execute API calls to fetch data needed for iteration

  2. Iteration Phase: Loop over a list (from setup or predefined)

  3. Generation Phase: Create one endpoint configuration per iteration item

  4. Execution Phase: Run the generated endpoints like normal static endpoints

The generated endpoints are combined with any static endpoints defined in the endpoints section.

Simple Iteration Over Static List

The simplest use case - iterate over a predefined list to create multiple similar endpoints.

Spec File (resources_api.yaml)

resources_api.yaml
name: "Resources API"

defaults:
  state:
    base_url: https://api.example.com/v1
  request:
    headers:
      Authorization: "Bearer {secrets.api_key}"

dynamic_endpoints:
  # Iterate over a static list of resource types
  - iterate: '["users", "orders", "products"]'
    into: "state.resource_type"

    endpoint:
      # Generate endpoint name from resource type
      name: "{state.resource_type}"
      description: "Fetch {state.resource_type} data"

      request:
        url: "{state.base_url}/{state.resource_type}"
        parameters:
          limit: 100
          offset: 0

      pagination:
        next_state:
          offset: "{state.offset + 100}"
        stop_condition: length(response.records) < 100

      response:
        records:
          jmespath: "data[]"
          primary_key: ["id"]

        processors:
          # Add resource type to each record for tracking
          - expression: "state.resource_type"
            output: "record._resource_type"

Using Replication

Running with Sling: sling run -r /path/to/replication.yaml

replication.yaml
source: MY_API
target: MY_TARGET_DB

streams:
  # The dynamic endpoints generate: users, orders, products
  '*':
    object: public.{stream_name}

This creates three endpoints from a single configuration, avoiding repetition.

Using Python

api_to_database.py
from sling import Replication, ReplicationStream

replication = Replication(
    source='MY_API',
    target='MY_TARGET_DB',
    streams={
        'users': ReplicationStream(object='public.users'),
        'orders': ReplicationStream(object='public.orders'),
        'products': ReplicationStream(object='public.products')
    }
)

replication.run()

Dynamic Discovery from API

Fetch the list of available resources from the API itself, then create endpoints dynamically.

Spec File (database_api.yaml)

database_api.yaml
name: "Database API"

defaults:
  state:
    base_url: https://api.database.com/v2
  request:
    headers:
      X-API-Key: "{secrets.api_key}"

dynamic_endpoints:
  - setup:
      # Fetch list of available tables from the API
      - request:
          url: "{state.base_url}/metadata/tables"
        response:
          processors:
            # Extract table names from response
            - expression: 'jmespath(response.json, "tables[].name")'
              output: "state.available_tables"
              aggregation: last

    # Create one endpoint per table discovered
    iterate: "state.available_tables"
    into: "state.table_name"

    endpoint:
      name: "table_{state.table_name}"
      description: "Data from {state.table_name} table"

      request:
        url: "{state.base_url}/tables/{state.table_name}/rows"
        parameters:
          limit: 1000
          offset: 0

      pagination:
        next_state:
          offset: "{state.offset + 1000}"
        stop_condition: length(response.records) < 1000

      response:
        records:
          jmespath: "rows[]"
          primary_key: ["id"]

        processors:
          # Tag each record with source table name
          - expression: "state.table_name"
            output: "record._source_table"

Using Replication

replication.yaml
source: MY_API
target: MY_TARGET_DB

# If API returns tables: ["customers", "invoices", "payments"]
# Then these endpoints are generated: table_customers, table_invoices, table_payments

streams:
  table_customers:
    object: staging.customers

  table_invoices:
    object: staging.invoices

  table_payments:
    object: staging.payments

Using Python

api_to_database.py
from sling import Replication, ReplicationStream

# The API will discover available tables automatically
replication = Replication(
    source='MY_API',
    target='MY_TARGET_DB',
    streams={
        'table_customers': ReplicationStream(object='staging.customers'),
        'table_invoices': ReplicationStream(object='staging.invoices'),
        'table_payments': ReplicationStream(object='staging.payments')
    }
)

replication.run()

Multi-Organization Endpoints

Create separate endpoints for each organization the authenticated user has access to.

Spec File (multi_org_api.yaml)

multi_org_api.yaml
name: "Multi-Organization API"

defaults:
  state:
    base_url: https://api.saas.com/v1
  request:
    headers:
      Authorization: "Bearer {secrets.access_token}"

dynamic_endpoints:
  - setup:
      # Fetch organizations the user can access
      - request:
          url: "{state.base_url}/user/organizations"
        response:
          processors:
            # Store the list of organizations
            - expression: 'jmespath(response.json, "organizations")'
              output: "state.org_list"
              aggregation: last

    # Create one endpoint per organization
    iterate: "state.org_list"
    into: "state.org"

    endpoint:
      # Use organization ID in endpoint name
      name: "org_{state.org.id}_events"
      description: "Events for {state.org.name}"

      request:
        url: "{state.base_url}/organizations/{state.org.id}/events"
        parameters:
          limit: 500
          cursor: "{state.cursor}"

      pagination:
        next_state:
          cursor: '{jmespath(response.json, "pagination.next_cursor")}'
        stop_condition: 'is_null(jmespath(response.json, "pagination.next_cursor"))'

      response:
        records:
          jmespath: "events[]"
          primary_key: ["event_id"]

        processors:
          # Add organization context to each record
          - expression: "state.org.id"
            output: "record.organization_id"
          - expression: "state.org.name"
            output: "record.organization_name"

Using Replication

replication.yaml
source: MY_API
target: MY_TARGET_DB

# If user has access to orgs 123 and 456, endpoints generated:
# org_123_events, org_456_events

streams:
  '*':
    object: public.{stream_name}

Geographic Regions with Metadata

Create endpoints for predefined geographic regions with additional metadata.

Spec File (regional_api.yaml)

regional_api.yaml
name: "Regional Sales API"

defaults:
  state:
    base_url: https://api.global-sales.com

    # Define regions with metadata
    regions:
      - code: "us-east"
        name: "US East Coast"
        timezone: "America/New_York"
      - code: "us-west"
        name: "US West Coast"
        timezone: "America/Los_Angeles"
      - code: "eu-west"
        name: "Europe West"
        timezone: "Europe/London"
      - code: "apac"
        name: "Asia Pacific"
        timezone: "Asia/Tokyo"

  request:
    headers:
      Authorization: "Bearer {secrets.api_key}"

dynamic_endpoints:
  # No setup needed - iterate over predefined regions
  - iterate: "state.regions"
    into: "state.region"

    endpoint:
      name: "sales_{state.region.code}"
      description: "Sales data for {state.region.name}"

      state:
        # Default to last 7 days
        start_date: '{date_format(date_add(now(), -7, "day"), "%Y-%m-%d")}'
        end_date: '{date_format(now(), "%Y-%m-%d")}'

      request:
        url: "{state.base_url}/regions/{state.region.code}/sales"
        parameters:
          date_from: "{state.start_date}"
          date_to: "{state.end_date}"
          page: "{state.page}"
          page_size: 250

      pagination:
        next_state:
          page: "{state.page + 1}"
        stop_condition: 'jmespath(response.json, "has_more") == false'

      response:
        records:
          jmespath: "sales[]"
          primary_key: ["sale_id"]

        processors:
          # Add region metadata to each record
          - expression: "state.region.code"
            output: "record.region_code"
          - expression: "state.region.name"
            output: "record.region_name"
          - expression: "state.region.timezone"
            output: "record.region_timezone"

Using Replication

replication.yaml
source: MY_API
target: MY_TARGET_DB

# Dynamic endpoints generated: sales_us-east, sales_us-west, sales_eu-west, sales_apac

streams:
  '*':
    object: public.{stream_name}

Combining Static and Dynamic Endpoints

Mix static endpoints (for unique resources) with dynamic endpoints (for similar resources).

Spec File (hybrid_api.yaml)

hybrid_api.yaml
name: "Hybrid API"

defaults:
  state:
    base_url: https://api.example.com/v1
  request:
    headers:
      Authorization: "Bearer {secrets.api_key}"

# Static endpoints for unique resources
endpoints:
  metadata:
    description: "API metadata and configuration"
    request:
      url: "{state.base_url}/metadata"
    response:
      records:
        jmespath: "."
        primary_key: ["api_version"]

  user_profile:
    description: "Authenticated user profile"
    request:
      url: "{state.base_url}/user/profile"
    response:
      records:
        jmespath: "."
        primary_key: ["user_id"]

# Dynamic endpoints for similar resources
dynamic_endpoints:
  - iterate: '["transactions", "accounts", "categories"]'
    into: "state.entity"

    endpoint:
      name: "{state.entity}"
      description: "Fetch {state.entity} data"

      request:
        url: "{state.base_url}/{state.entity}"
        parameters:
          limit: 100

      pagination:
        next_state:
          offset: "{state.offset + 100}"
        stop_condition: length(response.records) < 100

      response:
        records:
          jmespath: "data[]"
          primary_key: ["id"]

Using Replication

replication.yaml
source: MY_API
target: MY_TARGET_DB

# Total 5 endpoints: 2 static + 3 dynamic

streams:
  # Static endpoints
  metadata:
    object: config.api_metadata

  user_profile:
    object: config.user_profile

  # Dynamic endpoints
  '*':
    object: public.{stream_name}

Filtering During Setup

Only generate endpoints for resources matching specific criteria.

Spec File (filtered_api.yaml)

filtered_api.yaml
name: "Filtered Tables API"

defaults:
  state:
    base_url: https://api.example.com
  request:
    headers:
      X-API-Key: "{secrets.api_key}"

dynamic_endpoints:
  - setup:
      - request:
          url: "{state.base_url}/schemas/public/tables"
        response:
          processors:
            # Get all tables
            - expression: 'jmespath(response.json, "tables")'
              output: "state.all_tables"
              aggregation: last

            # Filter to only production tables (starting with "prod_")
            - expression: >
                filter(
                  state.all_tables,
                  "starts_with(name, 'prod_')"
                )
              output: "state.filtered_tables"

    # Only create endpoints for filtered tables
    iterate: "state.filtered_tables"
    into: "state.table"

    endpoint:
      name: "{state.table.name}"
      description: "Production table: {state.table.name}"

      request:
        url: "{state.base_url}/tables/{state.table.name}/data"
        parameters:
          limit: 1000

      pagination:
        next_state:
          offset: "{state.offset + 1000}"
        stop_condition: length(response.records) < 1000

      response:
        records:
          jmespath: "rows[]"
          primary_key: ["id"]

Using Replication

replication.yaml
source: MY_API
target: MY_TARGET_DB

# Only tables matching "prod_*" pattern are available

streams:
  prod_customers:
    object: public.customers

  prod_orders:
    object: public.orders

  # dev_* and test_* tables are excluded

Pattern Matching Dynamic Endpoints

Use wildcards to run multiple dynamic endpoints at once.

Using Replication with Wildcards

replication.yaml
source: MY_API
target: MY_TARGET_DB

# Run all organization endpoints for a specific org
streams:
  org_123_*:
    object: analytics.{stream}
    # This matches: org_123_events, org_123_users, etc.

  # Run all regional sales endpoints
  sales_*:
    object: sales.{stream}
    # This matches: sales_us-east, sales_us-west, etc.

Using Python with Pattern Matching

api_to_database.py
from sling import Replication, ReplicationStream

# You need to know the actual endpoint names or fetch them
replication = Replication(
    source='MY_API',
    target='MY_TARGET_DB',
    streams={
        # Run all org_123 endpoints
        'org_123_events': ReplicationStream(object='analytics.org_123_events'),
        'org_123_users': ReplicationStream(object='analytics.org_123_users'),
        'org_123_metrics': ReplicationStream(object='analytics.org_123_metrics'),
    }
)

replication.run()

Best Practices

1. Use Meaningful Endpoint Names

Generate descriptive names that clearly identify what each endpoint does:

# Good: Clear and descriptive
name: "region_{state.region.code}_daily_sales"

# Avoid: Generic or unclear
name: "endpoint_{state.index}"

2. Filter in Setup Phase

Only generate endpoints you'll actually use:

setup:
  - request:
      url: "{state.base_url}/resources"
    response:
      processors:
        # Filter to only active resources
        - expression: >
            filter(
              jmespath(response.json, "resources"),
              "status == 'active'"
            )
          output: "state.active_resources"

3. Add Context to Records

Use processors to tag records with metadata from the iteration:

processors:
  - expression: "state.org.id"
    output: "record.organization_id"
  - expression: "state.region.code"
    output: "record.region_code"

4. Validate Iteration Data

Ensure the data you're iterating over is valid:

setup:
  - request:
      url: "{state.base_url}/orgs"
    response:
      processors:
        - expression: 'jmespath(response.json, "organizations")'
          output: "state.orgs"
          aggregation: last

        # Log for debugging
        - expression: log("Found " + string(length(state.orgs)) + " organizations")
          output: ""

5. Use Static Endpoints for Unique Resources

Don't force everything into dynamic endpoints. Use static endpoints for one-off resources:

# Static for unique resources
endpoints:
  metadata:
    request:
      url: "{state.base_url}/metadata"

# Dynamic for similar resources
dynamic_endpoints:
  - iterate: "state.tables"
    into: "state.table"
    endpoint:
      name: "{state.table.name}"

6. Consider Performance

Be mindful of how many endpoints you generate:

# Be cautious with large lists
# Generating 1000+ endpoints can impact performance

setup:
  - request:
      url: "{state.base_url}/all_resources"
    response:
      processors:
        # Consider pagination or filtering large lists
        - expression: 'jmespath(response.json, "resources[0:100]")'
          output: "state.limited_resources"

Combining Dynamic Endpoints with Other Features

With Incremental Sync

dynamic_endpoints:
  - iterate: '["users", "orders", "products"]'
    into: "state.resource"

    endpoint:
      name: "{state.resource}"

      # Enable incremental sync for each endpoint
      sync: [last_updated_at]

      state:
        updated_since: >
          {coalesce(
            sync.last_updated_at,
            date_format(date_add(now(), -30, "day"), "%Y-%m-%dT%H:%M:%S%z")
          )}

      request:
        url: "{state.base_url}/{state.resource}"
        parameters:
          updated_since: "{state.updated_since}"

      response:
        records:
          jmespath: "data[]"
          primary_key: ["id"]

        processors:
          - expression: "record.updated_at"
            output: "state.last_updated_at"
            aggregation: "maximum"

With Backfill

dynamic_endpoints:
  - iterate: "state.regions"
    into: "state.region"

    endpoint:
      name: "region_{state.region.code}_events"

      sync: [last_date]

      iterate:
        # Support backfill with context.range_start/end
        over: >
          range(
            coalesce(context.range_start, sync.last_date, "2024-01-01"),
            coalesce(context.range_end, date_format(now(), "%Y-%m-%d")),
            "1d"
          )
        into: "state.current_date"

      request:
        url: "{state.base_url}/regions/{state.region.code}/events"
        parameters:
          date: "{state.current_date}"
PreviousBackfillNextUse Hooks Data

Last updated

Was this helpful?