Function create_batch

scrapegraphai/utils/batch_api.py:113–164

Create and submit an OpenAI Batch API job. Args: client: An initialized OpenAI client. requests: List of BatchRequest objects to submit. description: Human-readable description for the batch. Returns: The batch ID for tracking the job. Raises: V

(
    client: OpenAI,
    requests: List[BatchRequest],
    description: str = "ScrapeGraphAI batch scraping job",
)

def create_batch(
    client: OpenAI,
    requests: List[BatchRequest],
    description: str = "ScrapeGraphAI batch scraping job",
) -> str:
    """Create and submit an OpenAI Batch API job.

    Args:
        client: An initialized OpenAI client.
        requests: List of BatchRequest objects to submit.
        description: Human-readable description for the batch.

    Returns:
        The batch ID for tracking the job.

    Raises:
        ValueError: If the number of requests exceeds the API limit.
    """
    if len(requests) > MAX_REQUESTS_PER_BATCH:
        raise ValueError(
            f"Batch size {len(requests)} exceeds the maximum of "
            f"{MAX_REQUESTS_PER_BATCH}. Split into multiple batches."
        )

    # Build JSONL content
    jsonl_content = "\n".join(req.to_jsonl_line() for req in requests)

    logger.info(
        f"Uploading batch input file with {len(requests)} requests..."
    )

    # Upload the input file
    input_file = client.files.create(
        file=io.BytesIO(jsonl_content.encode("utf-8")),
        purpose="batch",
    )

    logger.info(f"Input file uploaded: {input_file.id}")

    # Create the batch
    batch = client.batches.create(
        input_file_id=input_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
        metadata={"description": description},
    )

    logger.info(
        f"Batch created: {batch.id} (status: {batch.status})"
    )

    return batch.id


def get_batch_status(client: OpenAI, batch_id: str) -> BatchJobInfo: