graflo.data_source.api

REST API data source implementation.

Runtime HTTP executor for API connectors. Configuration is built by :class:~graflo.architecture.contract.bindings.APIConnector.build_api_config from contract fields plus runtime credentials from a connection provider.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

APIConfig

Bases: ConfigBaseModel

Merged runtime configuration for REST API requests.

Built exclusively via :meth:APIConnector.build_api_config; not intended for direct construction in manifests or factory helpers.

Source code in graflo/data_source/api.py

class APIConfig(ConfigBaseModel):
    """Merged runtime configuration for REST API requests.

    Built exclusively via :meth:`APIConnector.build_api_config`; not intended
    for direct construction in manifests or factory helpers.
    """

    url: str
    method: str = "GET"
    headers: dict[str, str] = Field(default_factory=dict)
    auth: ApiAuth | None = None
    params: dict[str, Any] = Field(default_factory=dict)
    body: dict[str, Any] | str | None = None
    timeout: float | None = None
    retries: int = 0
    retry_backoff_factor: float = 0.1
    retry_status_forcelist: list[int] = Field(
        default_factory=lambda: [429, 500, 502, 503, 504]
    )
    verify: bool = True
    pagination: PaginationConfig | None = None
    row_annotations: dict[str, Any] = Field(default_factory=dict)
    watermark: WatermarkConfig | None = None

Attributes

auth = None class-attribute instance-attribute

body = None class-attribute instance-attribute

headers = Field(default_factory=dict) class-attribute instance-attribute

method = 'GET' class-attribute instance-attribute

pagination = None class-attribute instance-attribute

params = Field(default_factory=dict) class-attribute instance-attribute

retries = 0 class-attribute instance-attribute

retry_backoff_factor = 0.1 class-attribute instance-attribute

retry_status_forcelist = Field(default_factory=lambda: [429, 500, 502, 503, 504]) class-attribute instance-attribute

row_annotations = Field(default_factory=dict) class-attribute instance-attribute

timeout = None class-attribute instance-attribute

url instance-attribute

verify = True class-attribute instance-attribute

watermark = None class-attribute instance-attribute

APIDataSource

Bases: AbstractDataSource

Data source for REST API endpoints.

A request that fails after its retries raises requests.RequestException from :meth:iter_batches, on whichever page it happens.

With a declared watermark, :attr:high_water holds the largest watermark value among the rows yielded so far by the current iteration.

Source code in graflo/data_source/api.py

class APIDataSource(AbstractDataSource):
    """Data source for REST API endpoints.

    A request that fails after its retries raises ``requests.RequestException``
    from :meth:`iter_batches`, on whichever page it happens.

    With a declared watermark, :attr:`high_water` holds the largest watermark
    value among the rows yielded so far by the current iteration.
    """

    config: APIConfig
    source_type: DataSourceType = DataSourceType.API
    _high_water: Any = PrivateAttr(default=None)

    @property
    def high_water(self) -> Any:
        """Largest watermark value yielded so far; ``None`` without a watermark."""
        return self._high_water

    def _observe(self, row: dict[str, Any]) -> None:
        mark = self.config.watermark
        if mark is None:
            return
        value = get_at_path(row, mark.field)
        if value is None or isinstance(value, (dict, list)):
            return
        if self._high_water is None or later_watermark(value, self._high_water):
            self._high_water = value

    def _create_session(self) -> requests.Session:
        session = requests.Session()

        if self.config.retries > 0:
            retry_strategy = Retry(
                total=self.config.retries,
                backoff_factor=self.config.retry_backoff_factor,
                status_forcelist=self.config.retry_status_forcelist,
            )
            adapter = HTTPAdapter(max_retries=retry_strategy)
            session.mount("http://", adapter)
            session.mount("https://", adapter)

        auth = self.config.auth
        if auth is not None:
            if auth.auth_type == "basic":
                session.auth = HTTPBasicAuth(
                    auth.username or "",
                    auth.password or "",
                )
            elif auth.auth_type == "digest":
                session.auth = HTTPDigestAuth(
                    auth.username or "",
                    auth.password or "",
                )
            # No token, no header: a bare scheme or an empty key is not a
            # credential, and some servers reject it outright.
            elif auth.auth_type == "bearer" and auth.token:
                session.headers[auth.header_name] = (
                    f"{auth.prefix} {auth.token}".strip()
                )
            elif auth.auth_type == "api_key" and auth.token:
                session.headers[auth.header_name] = auth.token

        session.headers.update(self.config.headers)
        return session

    def iter_batches(
        self, batch_size: int = 1000, limit: int | None = None
    ) -> Iterator[list[dict]]:
        session = self._create_session()
        total_items = 0
        self._high_water = None
        resolved_response: ResolvedApiResponse | None = None

        try:
            pagination = self.config.pagination
            request = pagination.request if pagination else None
            offset = request.initial_offset if request else 0
            page = request.initial_page if request else 1
            cursor: str | None = request.initial_cursor if request else None
            carried: dict[str, Any] = {}
            carry_map = dict(request.carry_params) if request else {}

            while True:
                if limit is not None and total_items >= limit:
                    break

                params = dict(self.config.params)
                params.update(carried)

                page_limit = request.page_size if request else 0
                if request is not None and limit is not None:
                    page_limit = min(page_limit, limit - total_items)

                if request is not None:
                    if request.strategy == "offset":
                        params[request.offset_param] = offset
                        if request.limit_param is not None:
                            params[request.limit_param] = page_limit
                    elif request.strategy == "page":
                        params[request.page_param] = page
                        params[request.per_page_param] = page_limit
                    elif request.strategy == "cursor" and cursor is not None:
                        params[request.cursor_param] = cursor

                # A failed page raises: stopping here would hand the caller a
                # truncated stream that looks complete.
                body = self.config.body
                response = session.request(
                    method=self.config.method,
                    url=self.config.url,
                    params=params,
                    json=body if isinstance(body, dict) else None,
                    data=body if isinstance(body, str) else None,
                    timeout=self.config.timeout,
                    verify=self.config.verify,
                )
                response.raise_for_status()
                data = response.json()

                if pagination is not None and resolved_response is None:
                    resolved_response = ResolvedApiResponse.resolve(
                        pagination.response,
                        data,
                    )

                response_shape = (
                    resolved_response
                    if resolved_response is not None
                    else ResolvedApiResponse()
                )
                # Some APIs answer 200 with a failure in the body; reading it as
                # an empty page would end the stream as if it were complete.
                check_body(data, response_shape)
                items = extract_records(data, response_shape)
                batch_metadata = get_batch_metadata(data, response_shape)

                batch: list[dict] = []
                for item in items:
                    if limit is not None and total_items >= limit:
                        break
                    row = {**self.config.row_annotations, **batch_metadata, **item}
                    self._observe(row)
                    batch.append(row)
                    total_items += 1

                    if len(batch) >= batch_size:
                        yield batch
                        batch = []

                if batch:
                    yield batch

                if request is not None:
                    if not carry_map:
                        carry_map = detect_carry_params(data)
                        if carry_map:
                            logger.info(
                                "Auto-detected pagination carry_params: %s",
                                carry_map,
                            )
                    carried = update_carried_params(data, carry_map, carried)

                if limit is not None and total_items >= limit:
                    break

                if request is None:
                    break

                if not has_more_pages(
                    data,
                    response_shape,
                    items,
                    strategy=request.strategy,
                ):
                    break

                if request.strategy == "offset":
                    server_offset = next_offset_value(data, response_shape)
                    if server_offset is not None:
                        offset = server_offset
                    else:
                        offset += page_limit
                elif request.strategy == "page":
                    page += 1
                elif request.strategy == "cursor":
                    cursor = next_cursor_value(data, response_shape)
                    if not cursor:
                        break

        finally:
            session.close()

Attributes

config instance-attribute

high_water property

Largest watermark value yielded so far; None without a watermark.

source_type = DataSourceType.API class-attribute instance-attribute

Methods:

iter_batches(batch_size=1000, limit=None)

Source code in graflo/data_source/api.py

def iter_batches(
    self, batch_size: int = 1000, limit: int | None = None
) -> Iterator[list[dict]]:
    session = self._create_session()
    total_items = 0
    self._high_water = None
    resolved_response: ResolvedApiResponse | None = None

    try:
        pagination = self.config.pagination
        request = pagination.request if pagination else None
        offset = request.initial_offset if request else 0
        page = request.initial_page if request else 1
        cursor: str | None = request.initial_cursor if request else None
        carried: dict[str, Any] = {}
        carry_map = dict(request.carry_params) if request else {}

        while True:
            if limit is not None and total_items >= limit:
                break

            params = dict(self.config.params)
            params.update(carried)

            page_limit = request.page_size if request else 0
            if request is not None and limit is not None:
                page_limit = min(page_limit, limit - total_items)

            if request is not None:
                if request.strategy == "offset":
                    params[request.offset_param] = offset
                    if request.limit_param is not None:
                        params[request.limit_param] = page_limit
                elif request.strategy == "page":
                    params[request.page_param] = page
                    params[request.per_page_param] = page_limit
                elif request.strategy == "cursor" and cursor is not None:
                    params[request.cursor_param] = cursor

            # A failed page raises: stopping here would hand the caller a
            # truncated stream that looks complete.
            body = self.config.body
            response = session.request(
                method=self.config.method,
                url=self.config.url,
                params=params,
                json=body if isinstance(body, dict) else None,
                data=body if isinstance(body, str) else None,
                timeout=self.config.timeout,
                verify=self.config.verify,
            )
            response.raise_for_status()
            data = response.json()

            if pagination is not None and resolved_response is None:
                resolved_response = ResolvedApiResponse.resolve(
                    pagination.response,
                    data,
                )

            response_shape = (
                resolved_response
                if resolved_response is not None
                else ResolvedApiResponse()
            )
            # Some APIs answer 200 with a failure in the body; reading it as
            # an empty page would end the stream as if it were complete.
            check_body(data, response_shape)
            items = extract_records(data, response_shape)
            batch_metadata = get_batch_metadata(data, response_shape)

            batch: list[dict] = []
            for item in items:
                if limit is not None and total_items >= limit:
                    break
                row = {**self.config.row_annotations, **batch_metadata, **item}
                self._observe(row)
                batch.append(row)
                total_items += 1

                if len(batch) >= batch_size:
                    yield batch
                    batch = []

            if batch:
                yield batch

            if request is not None:
                if not carry_map:
                    carry_map = detect_carry_params(data)
                    if carry_map:
                        logger.info(
                            "Auto-detected pagination carry_params: %s",
                            carry_map,
                        )
                carried = update_carried_params(data, carry_map, carried)

            if limit is not None and total_items >= limit:
                break

            if request is None:
                break

            if not has_more_pages(
                data,
                response_shape,
                items,
                strategy=request.strategy,
            ):
                break

            if request.strategy == "offset":
                server_offset = next_offset_value(data, response_shape)
                if server_offset is not None:
                    offset = server_offset
                else:
                    offset += page_limit
            elif request.strategy == "page":
                page += 1
            elif request.strategy == "cursor":
                cursor = next_cursor_value(data, response_shape)
                if not cursor:
                    break

    finally:
        session.close()

DependentAPIDataSource

Bases: AbstractDataSource

One request per record of a parent source, its rows carrying the variables.

For each parent record, the variables declared in fields are read by dot path; a record missing any of them is skipped. They fill {variable} in the child's URL (percent-encoded), params and body, and every child row carries them as fields unless the row has a field of the same name.

Source code in graflo/data_source/api.py

class DependentAPIDataSource(AbstractDataSource):
    """One request per record of a parent source, its rows carrying the variables.

    For each parent record, the variables declared in *fields* are read by dot
    path; a record missing any of them is skipped. They fill ``{variable}`` in
    the child's URL (percent-encoded), params and body, and every child row
    carries them as fields unless the row has a field of the same name.
    """

    parent: AbstractDataSource
    config: APIConfig
    fields: dict[str, str]
    source_type: DataSourceType = DataSourceType.API
    _high_water: Any = PrivateAttr(default=None)

    @property
    def high_water(self) -> Any:
        """Largest watermark value over every child request of the current iteration."""
        return self._high_water

    def _values(self, record: dict[str, Any]) -> dict[str, Any] | None:
        values: dict[str, Any] = {}
        for variable, path in self.fields.items():
            value = get_at_path(record, path)
            if value is None or isinstance(value, (dict, list)):
                return None
            values[variable] = value
        return values

    def _request_for(self, values: dict[str, Any]) -> APIConfig:
        body = self.config.body
        return self.config.model_copy(
            update={
                "url": render_text(self.config.url, values, quote_values=True),
                "params": render_param(self.config.params, values),
                "body": None if body is None else render_body(body, values),
            }
        )

    def iter_batches(
        self, batch_size: int = 1000, limit: int | None = None
    ) -> Iterator[list[dict]]:
        self._high_water = None
        total = 0
        for parent_batch in self.parent.iter_batches(batch_size=batch_size):
            for record in parent_batch:
                if limit is not None and total >= limit:
                    return
                values = self._values(record)
                if values is None:
                    continue
                child = APIDataSource(config=self._request_for(values))
                remaining = None if limit is None else limit - total
                for batch in child.iter_batches(batch_size=batch_size, limit=remaining):
                    rows = [{**values, **row} for row in batch]
                    total += len(rows)
                    yield rows
                mark = child.high_water
                if mark is not None and (
                    self._high_water is None or later_watermark(mark, self._high_water)
                ):
                    self._high_water = mark

Attributes

config instance-attribute

fields instance-attribute

high_water property

Largest watermark value over every child request of the current iteration.

parent instance-attribute

source_type = DataSourceType.API class-attribute instance-attribute

Methods:

iter_batches(batch_size=1000, limit=None)

Source code in graflo/data_source/api.py

def iter_batches(
    self, batch_size: int = 1000, limit: int | None = None
) -> Iterator[list[dict]]:
    self._high_water = None
    total = 0
    for parent_batch in self.parent.iter_batches(batch_size=batch_size):
        for record in parent_batch:
            if limit is not None and total >= limit:
                return
            values = self._values(record)
            if values is None:
                continue
            child = APIDataSource(config=self._request_for(values))
            remaining = None if limit is None else limit - total
            for batch in child.iter_batches(batch_size=batch_size, limit=remaining):
                rows = [{**values, **row} for row in batch]
                total += len(rows)
                yield rows
            mark = child.high_water
            if mark is not None and (
                self._high_water is None or later_watermark(mark, self._high_water)
            ):
                self._high_water = mark

Functions: