Module scrapfly.batch

Streaming multipart/mixed parser for the POST /scrape/batch endpoint.

The API emits one part per scrape result as each scrape completes; the client must consume parts as they arrive (not after the whole response lands) to get the end-to-end streaming benefit.

Design notes: - Pure-Python parser (no new deps). Reuses requests streaming iter_content. - Works uniformly for JSON and msgpack part bodies — the negotiated part content-type is surfaced to the caller in the yielded tuple. - Does NOT perform decompression itself — requests already handles Content-Encoding gzip/zstd at the envelope level when stream=True is set with decode_content=True (default on requests Response).

Functions

def decode_part_body(headers: Dict[str, str], body: bytes, body_handler)

Expand source code

def decode_part_body(
    headers: Dict[str, str],
    body: bytes,
    body_handler,
):
    """
    Decode one part body according to its Content-Type header.
    Delegates to the existing ResponseBodyHandler for msgpack/json
    symmetry with single /scrape responses.
    """

    content_type = headers.get("content-type", "application/json")

    # body_handler.__call__ takes (content, content_type) and returns
    # a parsed dict. It handles both JSON and msgpack.
    return body_handler(content=body, content_type=content_type)

Decode one part body according to its Content-Type header. Delegates to the existing ResponseBodyHandler for msgpack/json symmetry with single /scrape responses.

def error_from_api_error_part(parsed: Dict, headers: Dict[str, str], request)

Expand source code

def error_from_api_error_part(parsed: Dict, headers: Dict[str, str], request):
    """Build the typed per-part error for an API-generated error body."""
    from .errors import ApiHttpClientError, ApiHttpServerError, ErrorFactory, api_error_args

    # A part has no HTTP status of its own, so the envelope decode falls back
    # to the status the multipart header carries for this part.
    args = api_error_args(parsed, http_status_code=_safe_int(headers.get("x-scrapfly-scrape-status"), 500))
    resource = args["resource"]
    http_code = args["http_status_code"]

    is_scraper_resource = resource in ErrorFactory.RESOURCE_TO_ERROR

    if http_code in ErrorFactory.HTTP_STATUS_TO_ERROR and not is_scraper_resource:
        error_class = ErrorFactory.HTTP_STATUS_TO_ERROR[http_code]
    elif is_scraper_resource:
        error_class = ErrorFactory.RESOURCE_TO_ERROR[resource]
    else:
        error_class = ApiHttpServerError if http_code >= 500 else ApiHttpClientError

    return error_class(
        request=request,
        response=_synthesize_part_response(request, http_code, parsed.get("reason")),
        **args,
    )

Build the typed per-part error for an API-generated error body.

def is_api_error_part(parsed, headers: Dict[str, str]) ‑> bool

Expand source code

def is_api_error_part(parsed, headers: Dict[str, str]) -> bool:
    """Detect an API-generated error part (status >= 400, no scrape envelope)."""

    if _safe_int(headers.get("x-scrapfly-scrape-status"), 0) < 400:
        return False

    return isinstance(parsed, dict) and "result" not in parsed and "config" not in parsed

Detect an API-generated error part (status >= 400, no scrape envelope).

def iter_batch_parts(response) ‑> Iterator[Tuple[Dict[str, str], bytes]]

Expand source code

def iter_batch_parts(
    response,  # requests.Response — duck-typed to avoid circular imports
) -> Iterator[Tuple[Dict[str, str], bytes]]:
    """
    Iterate (part_headers, part_body) tuples from a streaming
    multipart/mixed response. The per-part Content-Type is in
    `part_headers['content-type']` (lowercased key), and the
    correlation_id is in `part_headers['x-scrapfly-correlation-id']`.

    The caller is responsible for decoding `part_body` based on the
    part's Content-Type (JSON vs msgpack).

    Raises ValueError if the outer Content-Type is not multipart/mixed
    or if the boundary parameter is missing.
    """

    envelope_ct = response.headers.get("Content-Type", "")
    mime, params = _parse_content_type(envelope_ct)

    if mime != "multipart/mixed":
        raise ValueError(
            f"scrape_batch: expected Content-Type multipart/mixed, got {envelope_ct!r}"
        )

    boundary_str = params.get("boundary")

    if not boundary_str:
        raise ValueError(
            f"scrape_batch: Content-Type multipart/mixed is missing boundary parameter: {envelope_ct!r}"
        )

    boundary = boundary_str.encode("ascii")

    chunks = response.iter_content(chunk_size=8 * 1024)
    reader = _BufferedMultipartReader(chunks, boundary)

    # Skip anything before the first --boundary.
    reader.discard_prefix()

    while True:
        # After each --boundary we expect either CRLF (more parts) or
        # `--` (terminator). RFC 2046 mandates CRLF; any server
        # deviating from that is broken — return cleanly rather than
        # try to guess a framing variant.
        suffix = reader.read_exact(2)

        if suffix == b"--":
            # Final boundary. Drain CRLF and any epilogue.
            return

        if suffix != _CRLF:
            return

        # Read headers up to the blank line.
        header_block = reader.read_until(_CRLF + _CRLF)
        headers: Dict[str, str] = {}

        for line in header_block.split(_CRLF):
            if not line or b":" not in line:
                continue

            k, _, v = line.partition(b":")
            headers[k.decode("ascii", errors="replace").strip().lower()] = (
                v.decode("ascii", errors="replace").strip()
            )

        # Body framing: prefer Content-Length (we always emit it
        # server-side), fall back to boundary-delimited scan.
        cl_raw = headers.get("content-length")
        body: bytes

        if cl_raw and cl_raw.isdigit():
            body = reader.read_exact(int(cl_raw))
        else:
            # Read until next boundary marker. The "\r\n--<boundary>"
            # sequence is the canonical delimiter per RFC 2046.
            body = reader.read_until(_CRLF + b"--" + boundary)

        yield headers, body

        # If we used Content-Length, we still need to consume the
        # trailing "\r\n--<boundary>" that starts the next boundary.
        if cl_raw and cl_raw.isdigit():
            reader.read_until(_CRLF + b"--" + boundary)

Iterate (part_headers, part_body) tuples from a streaming multipart/mixed response. The per-part Content-Type is in part_headers['content-type'] (lowercased key), and the correlation_id is in part_headers['x-scrapfly-correlation-id'].

The caller is responsible for decoding part_body based on the part's Content-Type (JSON vs msgpack).

Raises ValueError if the outer Content-Type is not multipart/mixed or if the boundary parameter is missing.