Module

content_layer.sources.rest

RESTSource - Content source for REST APIs.

Fetches content from any REST API that returns JSON, with configurable field mappings for content and frontmatter.

Requires: pip install bengal[rest] (installs aiohttp)

View source
1 Class

Classes

RESTSource
Content source for REST APIs. Fetches content from any REST API that returns JSON. Supports: - Cus…
7

Content source for REST APIs.

Fetches content from any REST API that returns JSON. Supports:

  • Custom headers (with environment variable expansion)
  • Configurable field mappings for content and frontmatter
  • Pagination (link header or cursor-based)

Configuration:

url: str - API endpoint URL (required)
headers: dict - Request headers (optional, supports ${ENV_VAR})
content_field: str - JSON path to content (default: "content")
id_field: str - JSON path to ID (default: "id")
frontmatter_fields: dict - Mapping of frontmatter keys to JSON paths
items_path: str - JSON path to items array (default: auto-detect)
pagination: dict - Pagination config (optional)
    strategy: str - "link_header" or "cursor"
    cursor_field: str - Field containing next cursor
Inherits from ContentSource

Methods 2

fetch_all async
Fetch all content from the API. Handles pagination automatically if configured.
0 AsyncIterator[Conte… 
async def fetch_all(self) -> AsyncIterator[ContentEntry]

Fetch all content from the API.

Handles pagination automatically if configured.

Returns

AsyncIterator[ContentEntry]

fetch_one async
Fetch a single item by ID.
1 ContentEntry | None 
async def fetch_one(self, id: str) -> ContentEntry | None

Fetch a single item by ID.

Parameters 1
id str

Item identifier
Returns

ContentEntry | None —

ContentEntry if found, None otherwise

Internal Methods 5
__init__
Initialize REST source.
2 None 
def __init__(self, name: str, config: dict[str, Any]) -> None

Initialize REST source.

Parameters 2
name str

Source name
config dict[str, Any]

Configuration with 'url' required
_extract_items
Extract items array from response.
1 list[dict[str, Any]] 
def _extract_items(self, data: Any) -> list[dict[str, Any]]

Extract items array from response.

Parameters 1
data Any

JSON response data
Returns

list[dict[str, Any]] —

List of item dictionaries

_item_to_entry
Convert API item to ContentEntry.
1 ContentEntry | None 
def _item_to_entry(self, item: dict[str, Any]) -> ContentEntry | None

Convert API item to ContentEntry.

Parameters 1
item dict[str, Any]

Item dictionary from API
Returns

ContentEntry | None —

ContentEntry or None if content missing

_get_nested
Get nested value by dot-separated path.
2 Any 
def _get_nested(self, obj: Any, path: str) -> Any

Get nested value by dot-separated path.

Parameters 2
obj Any

Object to traverse
path str

Dot-separated path (e.g., "data.items.0.name")
Returns

Any —

Value at path or None

_get_next_url
Extract next page URL from response.
2 str | None 
def _get_next_url(self, data: dict[str, Any], response: aiohttp.ClientResponse) -> str | None

Extract next page URL from response.

Parameters 2
data dict[str, Any]

Response JSON
response aiohttp.ClientResponse

aiohttp response object
Returns

str | None —

Next page URL or None